Fullstack @ HRT
At Hudson River Trading, our Fullstack team bridges the gap between people and our highly automated systems. We build the complex interfaces and tools that let people navigate, control, and extend HRT’s vast trading infrastructure, as well as the data it produces. Running systems at this scale comes with no shortage of engineering challenges. This post explores one of those challenges and how we solved it.
We have built a suite of internal web applications that offer a real-time window into our trading environment. These tools provide our Run The Day (RTD) team and other teams throughout the company the ability to observe and respond to the dynamic intraday state of our trading operations. This suite, known collectively as the ‘Ops GUIs’, handle a large volume of complex structured data which updates at sub-second intervals. The data ranges from process statuses to market rejects to health check results. Alongside these fixed GUIs, we offer a WYSIWYG dashboard builder to support the varied data and application needs of different teams. These dashboards are composed of ‘widgets’ that typically summarize the information available in a full application.
Users often have their custom dashboard and some fixed GUIs open at the same time, and each GUI instance can load a non-trivial amount of data into memory.
How do we avoid duplicate loading and processing of data without overhauling our frontend stack (e.g. to try and implement a server-side framework or to change our trading systems to be frontend-aware)?
Since the advent of the SharedWorker interface in 2010, which enables a background thread to be shared across tabs, many web programmers consider this a solved problem. In theory, this is an ideal solution, but in practice, SharedWorkers introduce several development challenges:
- The only way to communicate with a SharedWorker is via a MessageChannel; this is a major deviation from the typical function calls we use in frontend code, especially for async functions or those that return multiple responses (as you’d expect from a streaming API).
- MessageChannels utilize the structured clone algorithm [1] which mean objects lose their prototype chain along with some other niche properties when passed between contexts.
- Since SharedWorkers aren’t globally available across all browsers, implementing a fallback for environments without SharedWorker support becomes complex and error-prone.
We designed our solution to allow developers to implement their business logic within a single ‘Worker’ class. This class is instantiated in two ways (boolean passed to the constructor): as a ‘Client’ in the GUI or as an ‘Oracle’ in a SharedWorker (and that’s “client” as in “client / server”, where the “Oracle” is like the “server”). The HRTWorker, depending on whether it’s in client or oracle-mode, handles all underlying communication logic, abstracting it away from the developer. In client-mode, HRTWorker proxies requests to a corresponding SharedWorker with an equivalent oracle-mode instance.
Essentially, developers only need to extend ‘HRTWorker’ and implement their business logic. Here is an example of a Worker that a developer might implement:
class MyWorker extends HRTWorker {
constructor(isOracleMode: boolean) {
super(isOracleMode)
}
@remoteExecute
async *processData(input: DataSet): AsyncGenerator<Progress, Result> {
yield { stage: 'parsing', percent: 0 }
const parsed = await this.parse(input)
yield { stage: 'analyzing', percent: 50 }
const analyzed = await this.analyze(parsed)
yield { stage: 'complete', percent: 100 }
return analyzed
}
}And to make use of it inside of a GUI:
// In the interest of brevity, we'll abstract away the need to pass a reference to the SharedWorker to the client-mode instances
const sharedWorker = new SharedWorker("myworker.js")
// false here because it's running in client-mode (i.e. in a web app)
const processor = new MyWorker(false)
for await (const progress of processor.processData(myData)) {
console.log(`${progress.stage}: ${progress.percent}%`)
}And the SharedWorker module:
// true because it's running in oracle-mode (i.e. in a SharedWorker)
new MyWorker(true)
// On construction, the HRTWorker (when running in oracle-mode) will hook into the SharedWorkerContext
// and start listening for messages so there's no need for us to do anything else here.Now that we’ve taken a look at how this pattern can be used, let’s dive into how HRTWorker is implemented. There are three significant components to the HRTWorker class:
- Connection Management
- Execution Protocol (‘Conversations’)
- Transparent Method Decoration
Let’s examine each component in detail.
Connection Management
MessagePorts are fundamental to any SharedWorker implementation; they are the connection between different browser contexts. A web application receives a port upon which it can talk to the SharedWorker and vice versa. The first thing that we need to be able to do is to manage the lifecycle of these MessagePorts.
The code below illustrates how the SharedWorker (oracle-mode) tracks the originating client for a message and can then reply to that client.
abstract class ConnectionManagerMixin {
private ports: Map<string, MessagePort> = new Map()
constructor(protected isOracleMode: boolean, protected port?: MessagePort) {
// The first example of HRTWorker behaving differently based on the
// mode it is instantiated in. In oracle-mode, we need to listen to the
// SharedWorkerContext onconnect events to detect when a client connects.
if (this.isOracleMode) {
self.onconnect = (event: MessageEvent) => {
this.registerChannel(event.ports[0])
}
} else {
if (!this.port) throw new Error("Client-mode requires a port")
this.port.onmessage = this.handleMessageFromOracle.bind(this)
this.port.start()
}
}
protected registerChannel(port: MessagePort): string {
// We'll want an ID to identify clients in logging
const id = generateId()
// Start the listener for this port
port.onmessage = (message) => {
// We'll implement this in our Execution Protocol section
this.handleMessageFromClient(id, message);
}
this.ports.set(id, port)
// Just a simple example of a welcome message, in reality we might want
// to send metadata about the client-mode instance (web app name, user etc.)
this.sendMessage(id, { type: 'welcome', id })
return id
}
protected sendMessage(clientId: string, message: any): void {
this.ports.get(clientId)?.postMessage(message)
}
}Execution Protocol (‘Conversations’)
Cross-context interactions are modeled as structured ‘conversations’ taking place over a multiplexed channel (the MessagePort for the SharedWorker).
interface Conversation {
id: string
method: string
isInitiator: boolean
stream: AsyncGenerator
cleanup: (() => void)[]
}
enum MessageType {
CALL = 'call',
RESPONSE = 'response',
COMPLETE = 'complete',
ERROR = 'error'
CANCEL = 'cancel'
}A typical conversation/execution follows this pattern:
Client Oracle
| |
| CALL(id, method, args) |
|-------------------------->|
| | (executes method)
| |
| RESPONSE(id, data1) |
|<--------------------------|
| RESPONSE(id, data2) |
|<--------------------------|
| |
| COMPLETE(id, finalResult) |
|<--------------------------|
| |
done doneThe oracle-mode HRTWorker automatically handles method execution. When a CALL message is received on the MessagePort by the SharedWorker it:
- Looks up the requested method on itself
- Executes it with the provided arguments
- Streams any yielded values as
RESPONSEmessages - Sends the final return value as
COMPLETE - Propagates any errors as
ERRORmessages
In HRTWorker, this looks something like:
/**
* This mixin provides the conversation management framework for when the
* HRTWorker is in oracle-mode.
*
* When instantiated in client-mode, none of the code here will ever be run
* (handleMessageFromClient is only called in oracle-mode per our ConnectionManagerMixin)
*/
abstract class OracleConversationManagerMixin {
// All of the active conversations
private conversations = new Set<string>();
/**
* Called onmessage by the ConnectionManagerMixin
*/
protected async handleMessageFromClient(clientID: string, event: MessageEvent) {
// Note that we expect/need conversationID to be UUIDs generated by the client
const { type, method, args, conversationID } = event.data
switch (type) {
case MessageType.CALL:
this.conversations.add(conversationID)
await this.executeMethod(clientID, method, args)
break
case MessageType.CANCEL:
this.conversations.delete(conversationID)
break
}
}
private async executeMethod(conversationId: string, methodName: string, args: any[]) {
try {
// Get the actual method to execute
const method = this[methodName as keyof this] as Function
if (!method) throw new Error(`Method ${methodName} not found`)
// Execute and handle streaming results
const result = method.apply(this, args)
if (result && typeof result.next === 'function') {
// Handle AsyncGenerator
for await (const value of result) {
if (!this.conversations.has(conversationId)) return; // Stop if conversation was cancelled
this.sendResponse(conversationId, value)
}
this.sendComplete(conversationId)
} else {
// Handle regular return value
this.sendComplete(conversationId, await result)
}
} catch (error) {
this.sendError(conversationId, error)
}
}
// Some helpful wrappers to create structured messages to send to the client
private sendResponse(id: string, data: any) {
this.sendMessage(id, {
type: MessageType.RESPONSE,
data
})
}
private sendComplete(id: string, data?: any) {
this.sendMessage(id, {
type: MessageType.COMPLETE,
data
})
}
private sendError(id: string, error: Error) {
this.sendMessage(id, {
type: MessageType.ERROR,
error: {
message: error.message,
stack: error.stack,
name: error.constructor.name
}
})
}
}However, this implementation still requires manual construction of CALL messages in the web app for each remote invocation. This boilerplate code detracts from the developer experience and introduces room for hard-to-spot errors, which brings us to the final component.
Transparent Method Decoration
The final piece of the puzzle uses typescript experimental decorators – which allow us to wrap a method and access descriptor properties of the method that we’re wrapping – to create a transparent interface:
function remoteExecute(target: any, methodName: string, descriptor: PropertyDescriptor) {
const originalMethod = descriptor.value
descriptor.value = function(...args: any[]) {
// If we're the executor (oracle), run locally
if (this.isOracleMode) {
return originalMethod.apply(this, args)
}
// If we're the client, initiate remote call
return this.call(methodName, args)
}
}Finally, we can implement a conversation management framework for client-mode which @remoteExecute can invoke. In the below example we can construct an abstraction over the underlying channel (and the remote oracle’s async generator) using promises that can be resolved when we receive RESPONSE messages from our remote oracle.
Note that MessagePorts have limitations on the objects that they can serialise between browsing contexts, [1] and in this public example we haven’t added any safeguards to detect/warn developers that their methods yield/return unserialisable types.
The resulting ClientConversationManagerMixin looks something like this, and since we can control our remoteExecute decorator (to disable the entire remote oracle system if we want), we can seamlessly enable and disable workers depending on whether they’re available or even just via a configuration option.
/**
* This mixin for HRTWorker provides client-mode instances the ability to
* handle conversations with a SharedWorker. The 'call' method is invoked
* by the remoteExecute decorator.
*/
abstract class ClientConversationManagerMixin {
private conversations = new Map<string, any>()
/**
* Called by the remoteExecute decorator in client-mode to execute a method
* with some arguments on the SharedWorker.
*/
async *call(method: string, args: any[]): AsyncGenerator {
// Generate a UUID
const conversationID = generateId()
// Create conversation tracking
let resolveNext: (value: any) => void
let rejectNext: (error: Error) => void
const conversation = {
nextPromise: new Promise((resolve, reject) => {
resolveNext = resolve
rejectNext = reject
}),
resolveNext,
rejectNext,
}
// We need to keep track of the resolve/reject functions for our promise
// since we'll be receiving responses/errors asynchronously via
// our MessagePort's onmessage (which calls handleMessageFromOracle)
this.conversations.set(conversationID, conversation)
// Send the call
this.messagePort.postMessage({
type: 'CALL',
conversationID,
method,
args,
})
try {
while (true) {
const message = await conversation.nextPromise
switch (message.type) {
// Method has yielded something via its asyncgenerator on the
// SharedWorker
case 'RESPONSE':
// Set up next promise, so we don't miss messages
conversation.nextPromise = new Promise((resolve, reject) => {
conversation.resolveNext = resolve
conversation.rejectNext = reject
})
// Yield to our web application
yield message.data
break
// Method has returned a value on the SharedWorker
case 'COMPLETE':
return message.data
// Method has thrown an error on the SharedWorker
case 'ERROR':
throw new Error(message.error.message)
// Something has broken in our protocol
default:
throw new Error(`Unknown message type: ${message.type}`)
}
}
} finally {
this.conversations.delete(conversationID)
}
}
protected cancel(conversationID: string) {
const conversation = this.conversations.get(conversationID)
if (conversation) {
this.messagePort.postMessage({
type: 'CANCEL',
conversationID,
})
conversation.rejectNext(new Error('Conversation cancelled'))
this.conversations.delete(conversationID)
}
}
private handleMessageFromOracle(event: MessageEvent) {
const { conversationID, ...message } = event.data
const conversation = this.conversations.get(conversationID)
if (conversation) {
conversation.resolveNext(message)
}
}
}By combining these components, HRTWorker provides a powerful, reusable way to manage complex communication between browser contexts. Naturally, we’ve only scratched the surface here. If you’re the kind of engineer who loves building robust solutions to challenging problems like this, our Fullstack team at HRT is hiring. Check out our job post here for more info.
