-
-
Notifications
You must be signed in to change notification settings - Fork 645
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Introducing Modular Asynchronous Reliable Queueing System (MarQS). Works in dev * Convert MarQS to using lua and dealing with concurrency * Simplified the timeout queue and current concurrency is now a set instead of a flat value (to support idempotency) * Implement task heartbeating and reconnect the background workers CLI when the websocket connection reconnects * Start adding internal telemetry support for the server * Get env vars to work in dev and implement prisma tracing in webapp * Cleanup telemetry and implement it in the consumer * Implement dequeuing a message from a parent shared queue * Implement a custom logger exporter instead of using console log exporter * Use node instead of shell for generating protocol buffer code * Propogate trace context into debug logs, and allow turning off logger exporter through env vars * Switch to using baselime for internal otel data * Make orgMember optional to fix type issues * Provide the CLI dev env vars through the CLI, don’t build dotenv into facade * Removed the logger import * Address Matt’s comments * Addressing more of Matt’s comments * Handle sending an execution after a websocket connection closes * Remove auth from the env attributes to prevent obfuscation
- Loading branch information
Showing
57 changed files
with
2,748 additions
and
746 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
95 changes: 95 additions & 0 deletions
95
apps/webapp/app/v3/authenticatedSocketConnection.server.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| import { | ||
| ZodMessageHandler, | ||
| ZodMessageSender, | ||
| clientWebsocketMessages, | ||
| serverWebsocketMessages, | ||
| } from "@trigger.dev/core/v3"; | ||
| import { Evt } from "evt"; | ||
| import { randomUUID } from "node:crypto"; | ||
| import { AuthenticatedEnvironment } from "~/services/apiAuth.server"; | ||
| import { logger } from "~/services/logger.server"; | ||
| import { EnvironmentQueueConsumer } from "./marqs/environmentQueueConsumer.server"; | ||
| import type { WebSocket, MessageEvent, CloseEvent, ErrorEvent } from "ws"; | ||
|
|
||
| export class AuthenticatedSocketConnection { | ||
| public id: string; | ||
| public onClose: Evt<CloseEvent> = new Evt(); | ||
|
|
||
| private _sender: ZodMessageSender<typeof serverWebsocketMessages>; | ||
| private _environmentConsumer: EnvironmentQueueConsumer; | ||
| private _messageHandler: ZodMessageHandler<typeof clientWebsocketMessages>; | ||
|
|
||
| constructor(public ws: WebSocket, public authenticatedEnv: AuthenticatedEnvironment) { | ||
| this.id = randomUUID(); | ||
|
|
||
| this._sender = new ZodMessageSender({ | ||
| schema: serverWebsocketMessages, | ||
| sender: async (message) => { | ||
| return new Promise((resolve, reject) => { | ||
| ws.send(JSON.stringify(message), {}, (err) => { | ||
| if (err) { | ||
| reject(err); | ||
| return; | ||
| } | ||
|
|
||
| resolve(); | ||
| }); | ||
| }); | ||
| }, | ||
| }); | ||
|
|
||
| this._environmentConsumer = new EnvironmentQueueConsumer(authenticatedEnv, this._sender); | ||
|
|
||
| ws.addEventListener("message", this.#handleMessage.bind(this)); | ||
| ws.addEventListener("close", this.#handleClose.bind(this)); | ||
| ws.addEventListener("error", this.#handleError.bind(this)); | ||
|
|
||
| this._messageHandler = new ZodMessageHandler({ | ||
| schema: clientWebsocketMessages, | ||
| messages: { | ||
| READY_FOR_TASKS: async (payload) => { | ||
| await this._environmentConsumer.registerBackgroundWorker(payload.backgroundWorkerId); | ||
| }, | ||
|
|
||
| BACKGROUND_WORKER_MESSAGE: async (payload) => { | ||
| switch (payload.data.type) { | ||
| case "TASK_RUN_COMPLETED": { | ||
| await this._environmentConsumer.taskRunCompleted( | ||
| payload.backgroundWorkerId, | ||
| payload.data.completion | ||
| ); | ||
| break; | ||
| } | ||
| case "TASK_HEARTBEAT": { | ||
| await this._environmentConsumer.taskHeartbeat( | ||
| payload.backgroundWorkerId, | ||
| payload.data.id | ||
| ); | ||
| break; | ||
| } | ||
| } | ||
| }, | ||
| }, | ||
| }); | ||
| } | ||
|
|
||
| async initialize() { | ||
| this._sender.send("SERVER_READY", { id: this.id }); | ||
| } | ||
|
|
||
| async #handleMessage(ev: MessageEvent) { | ||
| const data = JSON.parse(ev.data.toString()); | ||
|
|
||
| await this._messageHandler.handleMessage(data); | ||
| } | ||
|
|
||
| async #handleClose(ev: CloseEvent) { | ||
| await this._environmentConsumer.stop(); | ||
|
|
||
| this.onClose.post(ev); | ||
| } | ||
|
|
||
| async #handleError(ev: ErrorEvent) { | ||
| logger.error("Websocket error", { ev }); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.