[PAPI] Retry steams upon not receiving data for 10 seconds (#20172)

* [PAPI] Retry steams upon not receiving data for 10 seconds

* [PAPI] Retry steams upon not receiving data for 10 seconds (#20176)

* debug message

* Do not listen for ws updates globally and do not timeout for those calls

* Timeout signal as a getter

* use ff for disablement

* accesor fn => getter

* send empty initial data in papi for global ws watch requests

---------

Co-authored-by: Gero Posmyk-Leinemann <[email protected]>

Author: filiptronicek

components/dashboard/src/app/AppRoutes.tsx

@@ -33,7 +33,6 @@ import { Blocked } from "./Blocked";
 import { BlockedRepositories } from "../admin/BlockedRepositories";
 import PersonalAccessTokenCreateView from "../user-settings/PersonalAccessTokensCreateView";
 import { CreateWorkspacePage } from "../workspaces/CreateWorkspacePage";
-import { WebsocketClients } from "./WebsocketClients";
 import { BlockedEmailDomains } from "../admin/BlockedEmailDomains";
 import { AppNotifications } from "../AppNotifications";
 import { projectsPathInstallGitHubApp } from "../projects/projects.routes";
@@ -263,7 +262,6 @@ export const AppRoutes = () => {
                             />
                         </Switch>
                     </div>
-                    <WebsocketClients />
                 </Route>
             </Switch>
         </Route>

components/dashboard/src/app/WebsocketClients.tsx

@@ -41,6 +41,7 @@ import { VerificationService } from "@gitpod/public-api/lib/gitpod/v1/verificati
 import { JsonRpcInstallationClient } from "./json-rpc-installation-client";
 import { InstallationService } from "@gitpod/public-api/lib/gitpod/v1/installation_connect";
 import { JsonRpcUserClient } from "./json-rpc-user-client";
+import { Timeout } from "@gitpod/gitpod-protocol/lib/util/timeout";
 
 const transport = createConnectTransport({
     baseUrl: `${window.location.protocol}//${window.location.host}/public-api`,
@@ -273,14 +274,26 @@ export function stream<Response>(
     let backoff = BASE_BACKOFF;
     const abort = new AbortController();
     (async () => {
+        // Only timeout after 10 seconds with no data in some environments
+        const experiments = getExperimentsClient();
+        const enableTimeout = await experiments.getValueAsync("supervisor_check_ready_retry", false, {});
+
         while (!abort.signal.aborted) {
+            const connectionTimeout = new Timeout(10_000, () => enableTimeout);
             try {
+                connectionTimeout.start();
+                connectionTimeout.signal?.addEventListener("abort", () => {
+                    console.error("Connection timed out after no response for 10s");
+                });
+
                 for await (const response of factory({
-                    signal: abort.signal,
+                    signal: AbortSignal.any([abort.signal, connectionTimeout.signal!]),
                     // GCP timeout is 10 minutes, we timeout 3 mins earlier
                     // to avoid unknown network errors and reconnect gracefully
                     timeoutMs: 7 * 60 * 1000,
                 })) {
+                    connectionTimeout.clear(); // connection is alive now, clear timeout
+
                     backoff = BASE_BACKOFF;
                     cb(response);
                 }
@@ -302,6 +315,8 @@ export function stream<Response>(
                     backoff = Math.min(2 * backoff, MAX_BACKOFF);
                     console.error(e);
                 }
+            } finally {
+                connectionTimeout.clear();
             }
             const jitter = Math.random() * 0.3 * backoff;
             const delay = backoff + jitter;

components/dashboard/src/service/public-api.ts

@@ -24,14 +24,18 @@ import { BookOpen, Code } from "lucide-react";
 import { ReactComponent as GitpodStrokedSVG } from "../icons/gitpod-stroked.svg";
 import { isGitpodIo } from "../utils";
 import PersonalizedContent from "./PersonalizedContent";
+import { useListenToWorkspacesWSMessages as useListenToWorkspacesStatusUpdates } from "../data/workspaces/listen-to-workspace-ws-messages";
 
 const WorkspacesPage: FunctionComponent = () => {
     const [limit, setLimit] = useState(50);
     const [searchTerm, setSearchTerm] = useState("");
     const [showInactive, setShowInactive] = useState(false);
     const [deleteModalVisible, setDeleteModalVisible] = useState(false);
+
     const { data, isLoading } = useListWorkspacesQuery({ limit });
     const deleteInactiveWorkspaces = useDeleteInactiveWorkspacesMutation();
+    useListenToWorkspacesStatusUpdates();
+
     const { toast } = useToast();
 
     // Sort workspaces into active/inactive groups

components/dashboard/src/workspaces/Workspaces.tsx

@@ -16,44 +16,44 @@ export class TimeoutSpec {
         const timeout = new Timeout(1);
         timeout.start();
         await timeout.await();
-        expect(timeout.signal()?.aborted).to.be.true;
+        expect(timeout.signal?.aborted).to.be.true;
     }
 
     @test
     async testSimpleRunNotStarted() {
         const timeout = new Timeout(1);
         await timeout.await();
-        expect(timeout.signal()).to.be.undefined;
+        expect(timeout.signal).to.be.undefined;
     }
 
     @test
     async testRestart() {
         const timeout = new Timeout(20);
         timeout.start();
         await timeout.await();
-        expect(timeout.signal()?.aborted).to.be.true;
+        expect(timeout.signal?.aborted).to.be.true;
 
         timeout.restart();
-        expect(timeout.signal()).to.not.be.undefined;
-        expect(timeout.signal()?.aborted).to.be.false;
+        expect(timeout.signal).to.not.be.undefined;
+        expect(timeout.signal?.aborted).to.be.false;
         await timeout.await();
-        expect(timeout.signal()?.aborted).to.be.true;
+        expect(timeout.signal?.aborted).to.be.true;
     }
 
     @test
     async testClear() {
         const timeout = new Timeout(1000);
         timeout.restart();
         timeout.clear();
-        expect(timeout.signal()).to.be.undefined;
+        expect(timeout.signal).to.be.undefined;
     }
 
     @test
     async testAbortCondition() {
         const timeout = new Timeout(1, () => false); // will never trigger abort
         timeout.start();
         await new Promise((resolve) => setTimeout(resolve, 50));
-        expect(timeout.signal()).to.not.be.undefined;
-        expect(timeout.signal()?.aborted).to.be.false;
+        expect(timeout.signal).to.not.be.undefined;
+        expect(timeout.signal?.aborted).to.be.false;
     }
 }

components/gitpod-protocol/src/util/timeout.spec.ts

@@ -34,6 +34,9 @@ export class Timeout {
 
         const abortController = new AbortController();
         this._abortController = abortController;
+        if (this.timeout === Infinity) {
+            return;
+        }
         this._timer = setTimeout(() => {
             if (this.abortCondition && !this.abortCondition()) {
                 return;
@@ -74,7 +77,7 @@ export class Timeout {
     /**
      * @returns The AbortSignal of the current timeout, if one is active.
      */
-    public signal(): AbortSignal | undefined {
+    get signal() {
         return this._abortController?.signal;
     }
 }

components/gitpod-protocol/src/util/timeout.ts

@@ -1133,6 +1133,9 @@ export class WorkspaceService {
                 response.status = status;
                 yield response;
             }
+        } else {
+            // we initially send an empty status update to let clients know the connection is alive
+            yield {} as WatchWorkspaceStatusResponse;
         }
         const it = this.watchWorkspaceStatus(userId, opts);
         for await (const instance of it) {

components/server/src/workspace/workspace-service.ts

@@ -114,7 +114,7 @@ export class SupervisorServiceClient {
             });
             const response = await fetch(wsSupervisorStatusUrl.toString(), {
                 credentials: "include",
-                signal: timeout.signal(),
+                signal: timeout.signal,
             });
             let result;
             if (response.ok) {
@@ -140,11 +140,11 @@ export class SupervisorServiceClient {
 
             // we want to track this kind of errors, as they are on the critical path (of revealing the workspace)
             isError = true;
-            trackCheckReady({ aborted: timeout.signal()?.aborted }, e);
+            trackCheckReady({ aborted: timeout.signal?.aborted }, e);
         } finally {
             if (!isError) {
                 // make sure we don't track twice in case of an error
-                trackCheckReady({ aborted: timeout.signal()?.aborted });
+                trackCheckReady({ aborted: timeout.signal?.aborted });
             }
             timeout.clear();
         }