fix(frontend): stop opening a y-websocket connection on every compile

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 });
 

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", () => {

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<WorkflowCompilationResponse> {
+  private compile(logicalPlan: Pick<LogicalPlan, "operators" | "links">): Observable<WorkflowCompilationResponse> {
     // create a Logical Plan based on the workflow graph
     // remove unnecessary information for schema propagation.
     const body = {

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);
       });
     });
   });

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<string> {
+    return new Set(
+      this.workflowActionService
+        .getTexeraGraph()
+        .getAllOperatorIDs()
+        .filter(operatorID => this.validateOperator(operatorID).isValid)
     );
-
-    return new WorkflowGraph(validOperators, validLinks, texeraGraph.getAllCommentBoxes());
   }
 }