[server, et. al] Add repeat(cb, ms) as replacement for setInterval, and use it across meta components

Author: geropl

components/ee/payment-endpoint/src/github/subscription-reconciler.ts

@@ -19,6 +19,7 @@ import { Plans, Plan } from "@gitpod/gitpod-protocol/lib/plans";
 import * as Webhooks from '@octokit/webhooks';
 import fetch from "node-fetch";
 import { EntityManager } from 'typeorm';
+import { repeat } from "@gitpod/gitpod-protocol/lib/util/repeat";
 
 @injectable()
 export class GithubSubscriptionReconciler {
@@ -199,8 +200,8 @@ export class GithubSubscriptionReconciler {
     }
 
     public start() {
-        setInterval(() => this.reconciliationTasks.enqueue(() => this.reconcilePendingEvents()), 1 * 60 * 1000); // every one minute
-        setInterval(() => this.reconciliationTasks.enqueue(async () => {
+        repeat(() => this.reconciliationTasks.enqueue(() => this.reconcilePendingEvents()), 1 * 60 * 1000); // every one minute
+        repeat(() => this.reconciliationTasks.enqueue(async () => {
             try {
                 // it's important we reconcile the latest pending events first before attempting to interpret GitHub's information.
                 await this.reconcilePendingEvents();

components/gitpod-db/src/typeorm/deleted-entry-gc.ts

@@ -7,13 +7,17 @@
 import { injectable, inject } from "inversify";
 import { TypeORM } from "./typeorm";
 import { Config } from "../config";
+import { repeat } from "@gitpod/gitpod-protocol/lib/util/repeat";
+import { Disposable, DisposableCollection } from "@gitpod/gitpod-protocol";
 
 
 @injectable()
-export class DeletedEntryGC {
+export class DeletedEntryGC implements Disposable {
     @inject(TypeORM) protected readonly typeORM: TypeORM;
     @inject(Config) protected readonly config: Config;
 
+    protected readonly disposables = new DisposableCollection();
+
     public start() {
         const cfg = this.config.deletedEntryGCConfig;
         if (!cfg.enabled) {
@@ -22,9 +26,15 @@ export class DeletedEntryGC {
         }
 
         console.info(`Deleted Entries GC enabled (running every ${cfg.intervalMS/(60*1000)} minutes)`);
-        setInterval(() => {
-            this.gc().catch(e => console.error("error while removing deleted entries", e));
-        }, cfg.intervalMS);
+        this.disposables.push(
+            repeat(() => {
+                this.gc().catch(e => console.error("error while removing deleted entries", e));
+            }, cfg.intervalMS)
+        );
+    }
+
+    public dispose() {
+        this.disposables.dispose();
     }
 
     protected async gc() {

components/gitpod-protocol/src/util/garbage-collected-cache.ts

@@ -4,6 +4,8 @@
  * See License-AGPL.txt in the project root for license information.
  */
 
+import { repeat } from "./repeat";
+
 
 interface CacheEntry<T> {
     key: string;
@@ -46,7 +48,7 @@ export class GarbageCollectedCache<T> {
     }
 
     protected regularlyCollectGarbage() {
-        setInterval(() => this.collectGarbage(), this.gcIntervalSeconds * 1000);
+        repeat(() => this.collectGarbage(), this.gcIntervalSeconds * 1000);
     }
 
     protected collectGarbage() {

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

@@ -0,0 +1,45 @@
+/**
+ * Copyright (c) 2021 Gitpod GmbH. All rights reserved.
+ * Licensed under the GNU Affero General Public License (AGPL).
+ * See License-AGPL.txt in the project root for license information.
+ */
+
+import { Disposable } from "..";
+import { log } from "./logging";
+
+/**
+ * This intends to be a drop-in replacement for 'setInterval' implemented with a 'setTimeout' chain
+ * to ensure we're not creating more timeouts than we can process.
+ * @param op
+ * @param everyMilliseconds
+ * @returns
+ */
+export function repeat(op: () => Promise<void> | void, everyMilliseconds: number): Disposable {
+    let timer: NodeJS.Timeout | undefined = undefined;
+    let stopped = false;
+    const repeated = () => {
+        if (stopped) {
+            // in case we missed the clearTimeout i 'await'
+            return;
+        }
+
+        timer = setTimeout(async () => {
+            try {
+                await op();
+            } catch (err) {
+                // catch error here to
+                log.error(err);
+            }
+
+            repeated(); // chain ourselves - after the 'await'
+        }, everyMilliseconds);
+    };
+    repeated();
+
+    return Disposable.create(() => {
+        stopped = true;
+        if (timer) {
+            clearTimeout(timer);
+        }
+    });
+}

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

@@ -10,9 +10,10 @@ import { WorkspaceDB, TracedWorkspaceDB, DBWithTracing } from '@gitpod/gitpod-db
 import { v4 as uuidv4 } from 'uuid';
 import { HeadlessWorkspaceEvent } from '@gitpod/gitpod-protocol/lib/headless-workspace-log';
 import { log } from '@gitpod/gitpod-protocol/lib/util/logging';
-import { PrebuiltWorkspaceUpdatable, PrebuiltWorkspace, Disposable } from '@gitpod/gitpod-protocol';
+import { PrebuiltWorkspaceUpdatable, PrebuiltWorkspace, Disposable, DisposableCollection } from '@gitpod/gitpod-protocol';
 import { TraceContext } from '@gitpod/gitpod-protocol/lib/util/tracing';
 import { LocalMessageBroker } from '../../../src/messaging/local-message-broker';
+import { repeat } from "@gitpod/gitpod-protocol/lib/util/repeat";
 
 export interface CheckRunInfo {
     owner: string;
@@ -33,15 +34,18 @@ export class PrebuildStatusMaintainer implements Disposable {
     @inject(TracedWorkspaceDB) protected readonly workspaceDB: DBWithTracing<WorkspaceDB>;
     @inject(LocalMessageBroker) protected readonly localMessageBroker: LocalMessageBroker;
     protected githubApiProvider: AuthenticatedGithubProvider;
-    protected messagebusListener?: Disposable;
-    protected periodicChecker?: NodeJS.Timer;
+    protected readonly disposables = new DisposableCollection();
 
     start(githubApiProvider: AuthenticatedGithubProvider): void {
         // set github before registering the msgbus listener - otherwise an incoming message and the github set might race
         this.githubApiProvider = githubApiProvider;
 
-        this.messagebusListener = this.localMessageBroker.listenForPrebuildUpdatableEvents((ctx, msg) => this.handlePrebuildFinished(ctx, msg));
-        this.periodicChecker = setInterval(this.periodicUpdatableCheck.bind(this), 60 * 1000) as any as NodeJS.Timer;
+        this.disposables.push(
+            this.localMessageBroker.listenForPrebuildUpdatableEvents((ctx, msg) => this.handlePrebuildFinished(ctx, msg))
+        );
+        this.disposables.push(
+            repeat(this.periodicUpdatableCheck.bind(this), 60 * 1000)
+        );
         log.debug("prebuild updatatable status maintainer started");
     }
 
@@ -207,13 +211,6 @@ export class PrebuildStatusMaintainer implements Disposable {
     }
 
     dispose(): void {
-        if (this.messagebusListener) {
-            this.messagebusListener.dispose();
-            this.messagebusListener = undefined;
-        }
-        if (this.periodicChecker !== undefined) {
-            clearInterval(this.periodicChecker);
-            this.periodicChecker = undefined;
-        }
+        this.disposables.dispose();
     }
 }

components/server/ee/src/prebuilds/prebuilt-status-maintainer.ts

@@ -12,6 +12,7 @@ import { SafePromise } from "@gitpod/gitpod-protocol/lib/util/safe-promise";
 import { StorageClient } from "../../../src/storage/storage-client";
 import { ConsensusLeaderQorum } from "../../../src/consensus/consensus-leader-quorum";
 import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
+import { repeat } from "@gitpod/gitpod-protocol/lib/util/repeat";
 
 const SNAPSHOT_TIMEOUT_SECONDS = 60 * 30;
 const SNAPSHOT_POLL_INTERVAL_SECONDS = 5;
@@ -35,10 +36,7 @@ export class SnapshotService {
     protected readonly runningSnapshots: Map<string, Promise<void>> = new Map();
 
     public async start(): Promise<Disposable> {
-        const timer = setInterval(() => this.pickupAndDriveFromDbIfWeAreLeader().catch(log.error), SNAPSHOT_DB_POLL_INTERVAL_SECONDS * 1000);
-        return {
-            dispose: () => clearInterval(timer)
-        }
+        return repeat(() => this.pickupAndDriveFromDbIfWeAreLeader().catch(log.error), SNAPSHOT_DB_POLL_INTERVAL_SECONDS * 1000);
     }
 
     public async pickupAndDriveFromDbIfWeAreLeader() {

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

@@ -9,6 +9,7 @@ import { ConsensusLeaderMessenger, HeartbeatMessage, RequestVoteMessage, CastVot
 import { Disposable } from "@gitpod/gitpod-protocol";
 import { Deferred } from "@gitpod/gitpod-protocol/lib/util/deferred";
 import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
+import { repeat } from "@gitpod/gitpod-protocol/lib/util/repeat";
 
 @injectable()
 /* Implements the leader election mechanism of the Raft concensus algorithm:
@@ -19,7 +20,6 @@ import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
 export class ConsensusLeaderQorum implements Disposable {
     @inject(ConsensusLeaderMessenger) protected readonly messenger: ConsensusLeaderMessenger;
 
-    protected clock: NodeJS.Timeout | undefined;
     protected messages: RaftMessage[] = [];
     protected readonly disposables: Disposable[] = [];
     protected consensusAchieved = new Deferred<boolean>();
@@ -67,8 +67,7 @@ export class ConsensusLeaderQorum implements Disposable {
         // register with the messenger
         this.uid = await this.messenger.register();
 
-        this.clock = setInterval(() => this.beatClock().catch((err) => log.error("consensus beatClock", err)), this.clockPeriod);
-        this.disposables.push({ dispose: () => clearTimeout(this.clock!) });
+        this.disposables.push(repeat(() => this.beatClock().catch((err) => log.error("consensus beatClock", err)), this.clockPeriod));
         this.disposables.push(this.messenger.on("heartbeat", msg => this.messages.push(msg)));
         this.disposables.push(this.messenger.on("requestVote", msg => this.messages.push(msg)));
         this.disposables.push(this.messenger.on("castVote", msg => this.messages.push(msg)));

components/server/src/consensus/consensus-leader-quorum.ts

@@ -11,17 +11,19 @@ import * as express from 'express';
 import * as crypto from 'crypto';
 import { GitpodHostUrl } from '@gitpod/gitpod-protocol/lib/util/gitpod-host-url';
 import * as session from 'express-session';
+import { repeat } from '@gitpod/gitpod-protocol/lib/util/repeat';
 
 export const pingPong: WsRequestHandler = (ws, req, next) => {
     let pingSentTimer: any;
-    const timer = setInterval(() => {
+    const disposable = repeat(() => {
         if (ws.readyState !== ws.OPEN) {
             return;
         }
         // wait 10 secs for a pong
         pingSentTimer = setTimeout(() => {
             // Happens very often, we do not want to spam the logs here
             ws.terminate();
+            disposable.dispose();
         }, 10000);
         ws.ping();
     }, 30000)
@@ -35,7 +37,7 @@ export const pingPong: WsRequestHandler = (ws, req, next) => {
         ws.pong(data);
     });
     ws.on('close', () => {
-        clearInterval(timer);
+        disposable.dispose();
     })
     next();
 }

components/server/src/express-util.ts

@@ -8,6 +8,7 @@ import { Disposable, DisposableCollection, Emitter } from '@gitpod/gitpod-protoc
 import { filePathTelepresenceAware } from '@gitpod/gitpod-protocol/lib/env';
 import { IDEOptions } from '@gitpod/gitpod-protocol/lib/ide-protocol';
 import { log } from '@gitpod/gitpod-protocol/lib/util/logging';
+import { repeat } from '@gitpod/gitpod-protocol/lib/util/repeat';
 import * as Ajv from 'ajv';
 import * as cp from 'child_process';
 import * as crypto from 'crypto';
@@ -91,7 +92,7 @@ export class IDEConfigService {
         this.validate = this.ajv.compile(scheme);
         this.reconcile("initial");
         fs.watchFile(this.configPath, () => this.reconcile("file changed"));
-        setInterval(() => this.reconcile("interval"), 60 * 60 * 1000 /* 1 hour */);
+        repeat(() => this.reconcile("interval"), 60 * 60 * 1000 /* 1 hour */);
     }
 
     get config(): Promise<IDEConfig> {

components/server/src/ide-config.ts

@@ -8,29 +8,27 @@ import { injectable, inject } from "inversify";
 import * as express from 'express';
 import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
 import { OneTimeSecretDB, DBWithTracing, TracedOneTimeSecretDB } from "@gitpod/gitpod-db/lib";
-import { Disposable } from "@gitpod/gitpod-protocol";
+import { Disposable, DisposableCollection } from "@gitpod/gitpod-protocol";
 import * as opentracing from 'opentracing';
 import { TraceContext } from "@gitpod/gitpod-protocol/lib/util/tracing";
 import { Config } from "./config";
+import { repeat } from "@gitpod/gitpod-protocol/lib/util/repeat";
 
 @injectable()
 export class OneTimeSecretServer implements Disposable {
     @inject(Config) protected readonly config: Config;
     @inject(TracedOneTimeSecretDB) protected readonly oneTimeSecretDB: DBWithTracing<OneTimeSecretDB>;
 
-    protected pruneTimeout: NodeJS.Timeout | undefined;
+    protected readonly disposables = new DisposableCollection();
 
     public startPruningExpiredSecrets() {
-        this.pruneTimeout = setInterval(() => this.oneTimeSecretDB.trace({}).pruneExpired(), 5*60*1000);
+        this.disposables.push(
+            repeat(() => this.oneTimeSecretDB.trace({}).pruneExpired(), 5*60*1000)
+        );
     }
 
     dispose(): void {
-        if (!this.pruneTimeout) {
-            return;
-        }
-
-        clearInterval(this.pruneTimeout);
-        this.pruneTimeout = undefined;
+        this.disposables.dispose();
     }
 
     get apiRouter(): express.Router {

components/server/src/one-time-secret-server.ts

@@ -11,6 +11,7 @@ import { Disposable } from "@gitpod/gitpod-protocol";
 import { ConsensusLeaderQorum } from "../consensus/consensus-leader-quorum";
 import { TraceContext } from "@gitpod/gitpod-protocol/lib/util/tracing";
 import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
+import { repeat } from "@gitpod/gitpod-protocol/lib/util/repeat";
 
 @injectable()
 export class TokenGarbageCollector {
@@ -23,7 +24,7 @@ export class TokenGarbageCollector {
 
     public async start(intervalSeconds?: number): Promise<Disposable> {
         const intervalSecs = (intervalSeconds || TokenGarbageCollector.GC_CYCLE_INTERVAL_SECONDS);
-        const timer = setInterval(async () => {
+        return repeat(async () => {
             try {
                 if (await this.leaderQuorum.areWeLeader()) {
                     await this.collectExpiredTokenEntries();
@@ -32,9 +33,6 @@ export class TokenGarbageCollector {
                 log.error("token garbage collector", err);
             }
         }, intervalSecs * 1000);
-        return {
-            dispose: () => clearInterval(timer)
-        }
     }
 
     protected async collectExpiredTokenEntries() {

components/server/src/user/token-garbage-collector.ts

@@ -14,6 +14,7 @@ import * as opentracing from 'opentracing';
 import { TracedWorkspaceDB, DBWithTracing, WorkspaceDB } from "@gitpod/gitpod-db/lib";
 import { TraceContext } from "@gitpod/gitpod-protocol/lib/util/tracing";
 import { Config } from "../config";
+import { repeat } from "@gitpod/gitpod-protocol/lib/util/repeat";
 
 /**
  * The WorkspaceGarbageCollector has two tasks:
@@ -34,10 +35,7 @@ export class WorkspaceGarbageCollector {
                 dispose: () => {}
             }
         }
-        const timer = setInterval(async () => this.garbageCollectWorkspacesIfLeader(), 30 * 60 * 1000);
-        return {
-            dispose: () => clearInterval(timer)
-        }
+        return repeat(async () => this.garbageCollectWorkspacesIfLeader(), 30 * 60 * 1000);
     }
 
     public async garbageCollectWorkspacesIfLeader() {