fix: stop y-websocket connection leaks in shared editing and reap idle /rtc connections at the gateway

bin/k8s/templates/base/gateway/gateway-routes.yaml

@@ -76,13 +76,6 @@ spec:
       backendRefs:
         - name: access-control-service-svc
           port: 9096
-    - matches:
-        - path:
-            type: PathPrefix
-            value: /rtc
-      backendRefs:
-        - name: y-websocket-server-svc
-          port: 1234
     - matches:
         - path:
             type: PathPrefix
@@ -101,6 +94,31 @@ spec:
         - name: webserver-svc
           port: 8080
 ---
+# Shared-editing (y-websocket) route. Kept as its own HTTPRoute so a
+# BackendTrafficPolicy can target only the /rtc cluster (idle-connection reaping
+# + circuit-breaker headroom) without affecting the other static routes.
+# Longest-prefix matching still gives /rtc priority over the / catch-all above.
+apiVersion: gateway.networking.k8s.io/v1
+kind: HTTPRoute
+metadata:
+  name: {{ .Release.Name }}-shared-editing-route
+  namespace: {{ .Release.Namespace }}
+spec:
+  parentRefs:
+    - name: {{ .Release.Name }}-gateway
+  {{- if and .Values.gatewayConfig .Values.gatewayConfig.hostname }}
+  hostnames:
+    - {{ .Values.gatewayConfig.hostname }}
+  {{- end }}
+  rules:
+    - matches:
+        - path:
+            type: PathPrefix
+            value: /rtc
+      backendRefs:
+        - name: {{ .Values.yWebsocketServer.name }}-svc
+          port: 1234
+---
 apiVersion: gateway.networking.k8s.io/v1
 kind: HTTPRoute
 metadata:

bin/k8s/templates/base/shared-editing-server/shared-editing-server-backend-traffic-policy.yaml

@@ -0,0 +1,50 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Traffic policy for the /rtc shared-editing (y-websocket) cluster.
+# The y-websocket reference server is single-process and every browser tab holds
+# a long-lived WebSocket, so abandoned/leaked connections can pile up against the
+# gateway's per-cluster max_connections circuit breaker (Envoy default: 1024) and
+# starve legitimate realtime traffic. This policy reaps idle upstream connections
+# and raises the connection ceiling for headroom. It is defense-in-depth for the
+# client-side connection leak fixed in the frontend; keep both.
+{{- if .Values.yWebsocketServer.trafficPolicy }}
+apiVersion: gateway.envoyproxy.io/v1alpha1
+kind: BackendTrafficPolicy
+metadata:
+  name: {{ .Release.Name }}-shared-editing-traffic-policy
+  namespace: {{ .Release.Namespace }}
+spec:
+  targetRefs:
+    - group: gateway.networking.k8s.io
+      kind: HTTPRoute
+      name: {{ .Release.Name }}-shared-editing-route
+  {{- with .Values.yWebsocketServer.trafficPolicy.maxConnections }}
+  circuitBreaker:
+    # Raise the per-cluster max_connections above Envoy's 1024 default.
+    maxConnections: {{ . }}
+  {{- end }}
+  {{- with .Values.yWebsocketServer.trafficPolicy.connectionIdleTimeout }}
+  timeout:
+    http:
+      # Drop connections that sit with no active request for this long. Active
+      # co-editing exchanges awareness/cursor updates well within this window, so
+      # only genuinely idle (abandoned) connections are reaped. (Envoy Gateway
+      # exposes the idle timeout under timeout.http, not timeout.tcp.)
+      connectionIdleTimeout: {{ . | quote }}
+  {{- end }}
+{{- end }}

bin/k8s/values-development.yaml

@@ -357,6 +357,17 @@ yWebsocketServer:
   name: y-websocket-server
   replicaCount: 1
   image: texera/y-websocket-server:latest
+  # Gateway BackendTrafficPolicy for the /rtc shared-editing WebSocket cluster.
+  # Reaps idle upstream connections and raises the per-cluster connection ceiling
+  # so abandoned/leaked realtime sockets can't starve the gateway's connection cap.
+  # Set to null/false to omit the policy entirely.
+  trafficPolicy:
+    # Drop upstream connections idle (no bytes) for this long. Active co-editing
+    # sends awareness/cursor updates well within the window; only abandoned
+    # sockets are reaped. Envoy/Gateway-API duration string.
+    connectionIdleTimeout: "5m"
+    # Per-cluster max_connections circuit breaker (Envoy default is 1024).
+    maxConnections: 8192
 
 
 pythonLanguageServer:

bin/k8s/values.yaml

@@ -346,6 +346,17 @@ yWebsocketServer:
   name: y-websocket-server
   replicaCount: 1
   image: texera/y-websocket-server:latest
+  # Gateway BackendTrafficPolicy for the /rtc shared-editing WebSocket cluster.
+  # Reaps idle upstream connections and raises the per-cluster connection ceiling
+  # so abandoned/leaked realtime sockets can't starve the gateway's connection cap.
+  # Set to null/false to omit the policy entirely.
+  trafficPolicy:
+    # Drop upstream connections idle (no bytes) for this long. Active co-editing
+    # sends awareness/cursor updates well within the window; only abandoned
+    # sockets are reaped. Envoy/Gateway-API duration string.
+    connectionIdleTimeout: "5m"
+    # Per-cluster max_connections circuit breaker (Envoy default is 1024).
+    maxConnections: 8192
 
 
 pythonLanguageServer:

frontend/src/app/workspace/service/execute-workflow/mock-workflow-plan.ts

@@ -30,7 +30,9 @@ import { LogicalPlan } from "../../types/execute-workflow.interface";
 
 export const mockWorkflowPlan_scan_result: WorkflowGraph = new WorkflowGraph(
   [mockScanPredicate, mockResultPredicate],
-  [mockScanResultLink]
+  [mockScanResultLink],
+  [],
+  /* enableSharedEditing */ false
 );
 
 export const mockLogicalPlan_scan_result: LogicalPlan = {
@@ -64,7 +66,9 @@ export const mockLogicalPlan_scan_result: LogicalPlan = {
 
 export const mockWorkflowPlan_scan_sentiment_result: WorkflowGraph = new WorkflowGraph(
   [mockScanPredicate, mockSentimentPredicate, mockResultPredicate],
-  [mockScanSentimentLink, mockSentimentResultLink]
+  [mockScanSentimentLink, mockSentimentResultLink],
+  [],
+  /* enableSharedEditing */ false
 );
 
 export const mockLogicalPlan_scan_sentiment_result: LogicalPlan = {

frontend/src/app/workspace/service/validation/validation-workflow.service.ts

@@ -388,6 +388,13 @@ export class ValidationWorkflowService {
       link => validOperatorIds.has(link.source.operatorID) && validOperatorIds.has(link.target.operatorID)
     );
 
-    return new WorkflowGraph(validOperators, validLinks, texeraGraph.getAllCommentBoxes());
+    // This graph is a throwaway used only to build a logical plan for compilation/validation; it is read once and
+    // never destroyed. Pass enableSharedEditing=false so it does not open (and leak) a /rtc WebSocket per edit.
+    return new WorkflowGraph(
+      validOperators,
+      validLinks,
+      texeraGraph.getAllCommentBoxes(),
+      /* enableSharedEditing */ false
+    );
   }
 }

frontend/src/app/workspace/service/workflow-graph/model/shared-model.ts

@@ -38,7 +38,7 @@ import { YType } from "../../../types/shared-editing.interface";
  */
 export class SharedModel {
   public yDoc: Y.Doc = new Y.Doc();
-  public wsProvider: WebsocketProvider;
+  public wsProvider?: WebsocketProvider;
   public awareness: Awareness;
   public operatorIDMap: Y.Map<YType<OperatorPredicate>>;
   public commentBoxMap: Y.Map<YType<CommentBox>>;
@@ -56,11 +56,15 @@ export class SharedModel {
    * @param wid workflow ID number, used as part of the address for the shared-editing room.
    * @param user current (local) user info, used for initializing local awareness (user presence).
    * @param productionSharedEditingServer whether to use production shared editing server
+   * @param enableSharedEditing whether to join the shared-editing room over the network. When false, the model is
+   *   purely local: no {@link WebsocketProvider}/`/rtc` socket is opened and a standalone {@link Awareness} is used.
+   *   This is used for throwaway/validation graphs that must never consume a real-time connection.
    */
   constructor(
     public wid?: number,
     public user?: User,
-    private productionSharedEditingServer?: boolean
+    private productionSharedEditingServer?: boolean,
+    private enableSharedEditing: boolean = true
   ) {
     // Initialize Y-structures.
     this.debugState = this.yDoc.getMap("debugActions");
@@ -77,13 +81,17 @@ export class SharedModel {
       }
     );
 
-    // Generate editing room number.
-    const websocketUrl = this.getYWebSocketBaseUrl();
-    const suffix = wid ? `${wid}` : uuid();
-    this.wsProvider = new WebsocketProvider(websocketUrl, suffix, this.yDoc);
-
-    // Initialize local user awareness information.
-    this.awareness = this.wsProvider.awareness;
+    if (this.enableSharedEditing) {
+      // Generate editing room number and join the shared-editing room over the network.
+      const websocketUrl = this.getYWebSocketBaseUrl();
+      const suffix = wid ? `${wid}` : uuid();
+      this.wsProvider = new WebsocketProvider(websocketUrl, suffix, this.yDoc);
+      // Initialize local user awareness information from the provider's awareness.
+      this.awareness = this.wsProvider.awareness;
+    } else {
+      // Local-only: no network. A standalone awareness keeps clientId/awareness reads valid downstream.
+      this.awareness = new Awareness(this.yDoc);
+    }
     this.clientId = this.awareness.clientID.toString();
     if (this.user) {
       const userState: CoeditorState = {
@@ -126,10 +134,14 @@ export class SharedModel {
    * Destroys internal structures related to Yjs and quit the editing room.
    */
   public destroy(): void {
-    this.awareness.destroy();
     try {
-      if (this.wsProvider.shouldConnect && this.wsProvider.wsconnected) this.wsProvider.disconnect();
+      // Fully tear down the provider: destroy() clears the _checkInterval/_resyncInterval reconnect timers,
+      // removes the beforeunload handler, broadcasts awareness removal (peers drop our cursor), and closes the
+      // socket. Calling disconnect() alone (the previous behavior) leaves the reconnect timer running, so the
+      // provider is never garbage-collected and keeps re-opening a zombie /rtc socket for the life of the tab.
+      this.wsProvider?.destroy();
     } catch (e) {}
+    this.awareness.destroy();
     this.yDoc.destroy();
   }
 }

frontend/src/app/workspace/service/workflow-graph/model/workflow-action.service.ts

@@ -778,7 +778,7 @@ export class WorkflowActionService {
    * @param workflow
    */
   public setTempWorkflow(workflow: Workflow): void {
-    if (this.texeraGraph.sharedModel.wsProvider.shouldConnect) {
+    if (this.texeraGraph.sharedModel.wsProvider?.shouldConnect) {
       this.texeraGraph.sharedModel.wsProvider.disconnect();
     }
     this.tempWorkflow = workflow;
@@ -789,7 +789,7 @@ export class WorkflowActionService {
    */
   public resetTempWorkflow(): void {
     this.tempWorkflow = undefined;
-    this.texeraGraph.sharedModel.wsProvider.connect();
+    this.texeraGraph.sharedModel.wsProvider?.connect();
   }
 
   public getTempWorkflow(): Workflow | undefined {

frontend/src/app/workspace/service/workflow-graph/model/workflow-graph.ts

@@ -160,10 +160,13 @@ export class WorkflowGraph {
   constructor(
     operatorPredicates: OperatorPredicate[] = [],
     operatorLinks: OperatorLink[] = [],
-    commentBoxes: CommentBox[] = []
+    commentBoxes: CommentBox[] = [],
+    enableSharedEditing: boolean = true
   ) {
-    // Initialize sharedModel in constructor to ensure config is loaded
-    this.sharedModel = new SharedModel();
+    // Initialize sharedModel in constructor to ensure config is loaded.
+    // Throwaway graphs (e.g. validation/compilation graphs) pass enableSharedEditing=false so they never open a
+    // /rtc WebSocket. Otherwise every such short-lived graph would leak a self-reconnecting connection.
+    this.sharedModel = new SharedModel(undefined, undefined, undefined, enableSharedEditing);
 
     operatorPredicates.forEach(op => this.sharedModel.operatorIDMap.set(op.operatorID, createYTypeFromObject(op)));
     operatorLinks.forEach(link => this.sharedModel.operatorLinkMap.set(link.linkID, link));