diff --git a/bin/k8s/templates/base/gateway/gateway-routes.yaml b/bin/k8s/templates/base/gateway/gateway-routes.yaml index ac8096aa212..2b88883c016 100644 --- a/bin/k8s/templates/base/gateway/gateway-routes.yaml +++ b/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: diff --git a/bin/k8s/templates/base/shared-editing-server/shared-editing-server-backend-traffic-policy.yaml b/bin/k8s/templates/base/shared-editing-server/shared-editing-server-backend-traffic-policy.yaml new file mode 100644 index 00000000000..e6177206f9d --- /dev/null +++ b/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 }} diff --git a/bin/k8s/values-development.yaml b/bin/k8s/values-development.yaml index dc7078e4688..f4e310dacb6 100644 --- a/bin/k8s/values-development.yaml +++ b/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: diff --git a/bin/k8s/values.yaml b/bin/k8s/values.yaml index 64642b517a9..7ca543dcdb5 100644 --- a/bin/k8s/values.yaml +++ b/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: diff --git a/frontend/src/app/workspace/service/execute-workflow/mock-workflow-plan.ts b/frontend/src/app/workspace/service/execute-workflow/mock-workflow-plan.ts index c7222345b2c..c078413e7b4 100644 --- a/frontend/src/app/workspace/service/execute-workflow/mock-workflow-plan.ts +++ b/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 = { 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 92ebb88bb22..1b5615039ac 100644 --- a/frontend/src/app/workspace/service/validation/validation-workflow.service.ts +++ b/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 + ); } } diff --git a/frontend/src/app/workspace/service/workflow-graph/model/shared-model.ts b/frontend/src/app/workspace/service/workflow-graph/model/shared-model.ts index 0974426bcf5..6542c44f64f 100644 --- a/frontend/src/app/workspace/service/workflow-graph/model/shared-model.ts +++ b/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>; public commentBoxMap: Y.Map>; @@ -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(); } } diff --git a/frontend/src/app/workspace/service/workflow-graph/model/workflow-action.service.ts b/frontend/src/app/workspace/service/workflow-graph/model/workflow-action.service.ts index e3ea66c024c..0b6b1a25cc5 100644 --- a/frontend/src/app/workspace/service/workflow-graph/model/workflow-action.service.ts +++ b/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 { diff --git a/frontend/src/app/workspace/service/workflow-graph/model/workflow-graph.ts b/frontend/src/app/workspace/service/workflow-graph/model/workflow-graph.ts index 4d4bb93edeb..a27a317cc90 100644 --- a/frontend/src/app/workspace/service/workflow-graph/model/workflow-graph.ts +++ b/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));