From 2973287fcd66d973a4df5c6317c7c07a16d0a72a Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Tue, 27 Oct 2026 16:35:24 -0700 Subject: [PATCH] fix(frontend): stop opening a y-websocket connection on every compile Each debounced compile planned a filtered copy of the graph, built by ValidationWorkflowService.getValidTexeraGraph() as a new WorkflowGraph. Its SharedModel opened a WebsocketProvider to a fresh rtc/ room that nothing ever closed, so every edit left one more socket retrying forever. The compile now plans the live graph and keeps only the valid operators and the links between them; getValidTexeraGraph() becomes getValidOperatorIDs(). The compile subscription also stops when the service is destroyed. Co-Authored-By: Claude Opus 5.5 --- .../component/menu/menu.component.spec.ts | 2 +- .../workflow-compiling.service.spec.ts | 52 ++++++++++++++++++- .../workflow-compiling.service.ts | 23 +++++--- .../validation-workflow.service.spec.ts | 16 +++--- .../validation/validation-workflow.service.ts | 30 ++++------- 5 files changed, 84 insertions(+), 39 deletions(-) diff --git a/frontend/src/app/workspace/component/menu/menu.component.spec.ts b/frontend/src/app/workspace/component/menu/menu.component.spec.ts index 3a5f8440b80..b7e6884819c 100644 --- a/frontend/src/app/workspace/component/menu/menu.component.spec.ts +++ b/frontend/src/app/workspace/component/menu/menu.component.spec.ts @@ -752,7 +752,7 @@ describe("MenuComponent", () => { const convert = vi .spyOn(component["workflowToPythonService"], "convertToPython") .mockReturnValue(of({ type: "success", pythonCode: "print(1)" } as any)); - const validSubgraph = vi.spyOn(validationWorkflowService, "getValidTexeraGraph"); + const validSubgraph = vi.spyOn(validationWorkflowService, "getValidOperatorIDs"); workflowActionService.addOperator(mockScanPredicate, mockPoint); validationStream$.next({ errors: {}, workflowEmpty: false }); diff --git a/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.spec.ts b/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.spec.ts index 28f975540e1..d73589c3e6d 100644 --- a/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.spec.ts +++ b/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.spec.ts @@ -649,7 +649,7 @@ describe("WorkflowCompilingService compile pipeline", () => { workflowActionService.addOperator(mockScanPredicate, mockPoint); workflowActionService.addOperator(mockResultPredicate, mockPoint); workflowActionService.addLink(mockScanResultLink); - // ScanSource requires `tableName`; without it the operator is filtered out of the valid graph + // ScanSource requires `tableName`; without it the operator is filtered out of the logical plan workflowActionService.setOperatorProperty(mockScanPredicate.operatorID, { tableName: "twitter" }); tick(WORKFLOW_COMPILATION_DEBOUNCE_TIME_MS); return httpTestingController.expectOne(compileUrl); @@ -743,6 +743,56 @@ describe("WorkflowCompilingService compile pipeline", () => { expect(service.getWorkflowCompilationState()).toBe(CompilationState.Succeeded); warn.mockRestore(); })); + + it("leaves an invalid operator, and the links touching it, out of the logical plan", fakeAsync(() => { + workflowActionService.addOperator(mockScanPredicate, mockPoint); + workflowActionService.addOperator(mockResultPredicate, mockPoint); + workflowActionService.addLink(mockScanResultLink); + // no tableName, which ScanSource requires + tick(WORKFLOW_COMPILATION_DEBOUNCE_TIME_MS); + const request = httpTestingController.expectOne(compileUrl); + + const body = JSON.parse(request.request.body); + expect(body.operators.map((operator: any) => operator.operatorID)).toEqual([mockResultPredicate.operatorID]); + expect(body.links).toEqual([]); + + request.flush({ operatorOutputSchemas: {}, operatorErrors: {} }); + })); + + // Compiling used to plan a filtered copy of the graph: a new WorkflowGraph, whose SharedModel opened a + // y-websocket connection to a room of its own that nothing ever closed, one more on every debounced edit. + it("opens no websocket when it compiles", fakeAsync(() => { + const openedUrls: string[] = []; + vi.stubGlobal( + "WebSocket", + class { + constructor(url: string) { + openedUrls.push(url); + } + } + ); + try { + buildWorkflowAndCompile().flush({ operatorOutputSchemas: {}, operatorErrors: {} }); + workflowActionService.setOperatorProperty(mockScanPredicate.operatorID, { tableName: "reddit" }); + tick(WORKFLOW_COMPILATION_DEBOUNCE_TIME_MS); + httpTestingController.expectOne(compileUrl).flush({ operatorOutputSchemas: {}, operatorErrors: {} }); + } finally { + vi.unstubAllGlobals(); + } + + expect(openedUrls).toEqual([]); + })); + + it("stops compiling once the service is destroyed", fakeAsync(() => { + buildWorkflowAndCompile().flush({ operatorOutputSchemas: {}, operatorErrors: {} }); + + // what the injector calls on teardown; @UntilDestroy() adds it to the class + (service as any).ngOnDestroy(); + workflowActionService.setOperatorProperty(mockScanPredicate.operatorID, { tableName: "reddit" }); + tick(WORKFLOW_COMPILATION_DEBOUNCE_TIME_MS); + + httpTestingController.expectNone(compileUrl); + })); }); describe("WorkflowCompilingService.applySchemaPropagationResult without a propagated input schema", () => { diff --git a/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.ts b/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.ts index dae47a0088b..07820f2e5cf 100644 --- a/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.ts +++ b/frontend/src/app/workspace/service/compile-workflow/workflow-compiling.service.ts @@ -28,6 +28,7 @@ import { ExecuteWorkflowService } from "../execute-workflow/execute-workflow.ser import { WorkflowActionService } from "../workflow-graph/model/workflow-action.service"; import { catchError, debounceTime, mergeMap } from "rxjs/operators"; import { DynamicSchemaService } from "../dynamic-schema/dynamic-schema.service"; +import { UntilDestroy, untilDestroyed } from "@ngneat/until-destroy"; import { AttributeType, CompilationState, @@ -61,6 +62,7 @@ export const WORKFLOW_COMPILATION_DEBOUNCE_TIME_MS = 500; * Specifically for schema autocomplete, by contract, property name `attribute` and `attributes` indicate the field is a column of the operator's input, * and schema propagation can provide autocomplete for the column names. */ +@UntilDestroy() @Injectable({ providedIn: "root", }) @@ -95,12 +97,19 @@ export class WorkflowCompilingService { .pipe(debounceTime(WORKFLOW_COMPILATION_DEBOUNCE_TIME_MS)) .pipe( mergeMap(() => { - const logicalPlan = ExecuteWorkflowService.getLogicalPlanRequest( - this.validationWorkflowService.getValidTexeraGraph(), - undefined - ); - return this.compile(logicalPlan); - }) + // Plan the live graph, then keep its valid operators and the links between them. Planning a filtered + // copy of the graph instead builds a new WorkflowGraph per compile, and its SharedModel opens a + // y-websocket connection that nothing ever closes. + const logicalPlan = ExecuteWorkflowService.getLogicalPlanRequest(this.workflowActionService.getTexeraGraph()); + const validOperatorIDs = this.validationWorkflowService.getValidOperatorIDs(); + return this.compile({ + operators: logicalPlan.operators.filter(operator => validOperatorIDs.has(operator.operatorID)), + links: logicalPlan.links.filter( + link => validOperatorIDs.has(link.fromOpId) && validOperatorIDs.has(link.toOpId) + ), + }); + }), + untilDestroyed(this) ) .subscribe(response => { if (response.physicalPlan) { @@ -299,7 +308,7 @@ export class WorkflowCompilingService { * that users can easily set the properties of the next operator. For eg: If there are two operators Source:Scan and KeywordSearch and * a link is created between them, the attributed of the table selected in Source can be propagated to the KeywordSearch operator. */ - private compile(logicalPlan: LogicalPlan): Observable { + private compile(logicalPlan: Pick): Observable { // create a Logical Plan based on the workflow graph // remove unnecessary information for schema propagation. const body = { diff --git a/frontend/src/app/workspace/service/validation/validation-workflow.service.spec.ts b/frontend/src/app/workspace/service/validation/validation-workflow.service.spec.ts index f0bee8f1c4e..f9dc285ef0c 100644 --- a/frontend/src/app/workspace/service/validation/validation-workflow.service.spec.ts +++ b/frontend/src/app/workspace/service/validation/validation-workflow.service.spec.ts @@ -555,30 +555,28 @@ describe("ValidationWorkflowService", () => { }); }); - describe("the graph sent to the compiler", () => { + describe("the operators sent to the compiler", () => { it("keeps an operator holding a reference inside a block, before any compile", () => { addBlockAround({ limit: "$K" }); - const validGraph = validationWorkflowService.getValidTexeraGraph(); - expect(validGraph.hasOperator(body.operatorID)).toBe(true); - expect(validGraph.getOperator(body.operatorID).operatorProperties).toEqual({ limit: "$K" }); + expect(validationWorkflowService.getValidOperatorIDs().has(body.operatorID)).toBe(true); // a name no enclosing Loop Start declares is no error either workflowActionservice.setOperatorProperty(body.operatorID, { limit: "$foo" }); - expect(validationWorkflowService.getValidTexeraGraph().hasOperator(body.operatorID)).toBe(true); + expect(validationWorkflowService.getValidOperatorIDs().has(body.operatorID)).toBe(true); }); it("leaves out one holding a reference outside every block, or with any other error", () => { addBlockAround({ limit: "$K", fraction: "warm" }); - expect(validationWorkflowService.getValidTexeraGraph().hasOperator(body.operatorID)).toBe(false); + expect(validationWorkflowService.getValidOperatorIDs().has(body.operatorID)).toBe(false); // a reference in an array property is an error everywhere workflowActionservice.setOperatorProperty(body.operatorID, { limit: 1, columns: "$K" }); - expect(validationWorkflowService.getValidTexeraGraph().hasOperator(body.operatorID)).toBe(false); + expect(validationWorkflowService.getValidOperatorIDs().has(body.operatorID)).toBe(false); // outside every block, a reference in an integer property is a type error workflowActionservice.setOperatorProperty(body.operatorID, { limit: "$K" }); workflowActionservice.deleteLinkWithID(mockScalaExecutorLoopEndLink.linkID); - expect(validationWorkflowService.getValidTexeraGraph().hasOperator(body.operatorID)).toBe(false); + expect(validationWorkflowService.getValidOperatorIDs().has(body.operatorID)).toBe(false); // and a valid operator stays in workflowActionservice.setOperatorProperty(body.operatorID, { limit: 1 }); - expect(validationWorkflowService.getValidTexeraGraph().hasOperator(body.operatorID)).toBe(true); + expect(validationWorkflowService.getValidOperatorIDs().has(body.operatorID)).toBe(true); }); }); }); diff --git a/frontend/src/app/workspace/service/validation/validation-workflow.service.ts b/frontend/src/app/workspace/service/validation/validation-workflow.service.ts index fb586513ba1..95c0c47b9e1 100644 --- a/frontend/src/app/workspace/service/validation/validation-workflow.service.ts +++ b/frontend/src/app/workspace/service/validation/validation-workflow.service.ts @@ -26,7 +26,6 @@ import Ajv, { ErrorObject } from "ajv"; import { map } from "rxjs/operators"; import { DynamicSchemaService } from "../dynamic-schema/dynamic-schema.service"; import { UntilDestroy, untilDestroyed } from "@ngneat/until-destroy"; -import { WorkflowGraph, WorkflowGraphReadonly } from "../workflow-graph/model/workflow-graph"; import { collectReferences, isReference, @@ -414,28 +413,17 @@ export class ValidationWorkflowService { } /** - * Gets a filtered version of the TexeraGraph containing only valid operators and their corresponding links. - * This method will create a copy of the TexeraGraph and do the validation on top of it. + * Gets the IDs of the operators in the TexeraGraph that pass validation, validating each one afresh. + * The compiler is sent these operators and the links between them. * - * @returns A json-schema-wise valid TexeraGraph + * @returns the IDs of the json-schema-wise valid operators */ - public getValidTexeraGraph(): WorkflowGraphReadonly { - const texeraGraph = this.workflowActionService.getTexeraGraph(); - const allOperators = texeraGraph.getAllOperators(); - const allLinks = texeraGraph.getAllLinks(); - - // Filter valid operators using validation service - const validOperators = allOperators.filter(operator => { - const validation = this.validateOperator(operator.operatorID); - return validation.isValid; - }); - - // Filter links to only include those connecting valid operators - const validOperatorIds = new Set(validOperators.map(op => op.operatorID)); - const validLinks = allLinks.filter( - link => validOperatorIds.has(link.source.operatorID) && validOperatorIds.has(link.target.operatorID) + public getValidOperatorIDs(): ReadonlySet { + return new Set( + this.workflowActionService + .getTexeraGraph() + .getAllOperatorIDs() + .filter(operatorID => this.validateOperator(operatorID).isValid) ); - - return new WorkflowGraph(validOperators, validLinks, texeraGraph.getAllCommentBoxes()); } }