Scheduler abstraction (#2829)

src/lib/routes/admin-api/config.test.ts

@@ -31,7 +31,6 @@ async function getSetup() {
         destroy: () => {
             services.versionService.destroy();
             services.clientInstanceService.destroy();
-            services.apiTokenService.destroy();
         },
     };
 }

src/lib/routes/admin-api/context.test.ts

@@ -23,7 +23,6 @@ async function getSetup() {
         destroy: () => {
             services.versionService.destroy();
             services.clientInstanceService.destroy();
-            services.apiTokenService.destroy();
         },
     };
 }

src/lib/routes/admin-api/metrics.test.ts

@@ -25,7 +25,6 @@ async function getSetup() {
         destroy: () => {
             services.versionService.destroy();
             services.clientInstanceService.destroy();
-            services.apiTokenService.destroy();
         },
     };
 }

src/lib/routes/admin-api/strategy.test.ts

@@ -21,7 +21,6 @@ async function getSetup() {
     destroy = () => {
         services.versionService.destroy();
         services.clientInstanceService.destroy();
-        services.apiTokenService.destroy();
     };
 
     return {

src/lib/routes/admin-api/tag.test.ts

@@ -24,7 +24,6 @@ async function getSetup() {
         destroy: () => {
             services.versionService.destroy();
             services.clientInstanceService.destroy();
-            services.apiTokenService.destroy();
         },
     };
 }

src/lib/routes/backstage.test.ts

@@ -21,5 +21,4 @@ test('should enable prometheus', async () => {
         .expect(200);
     services.versionService.destroy();
     services.clientInstanceService.destroy();
-    services.apiTokenService.destroy();
 });

src/lib/routes/client-api/feature.test.ts

@@ -26,7 +26,6 @@ async function getSetup() {
         destroy: () => {
             services.versionService.destroy();
             services.clientInstanceService.destroy();
-            services.apiTokenService.destroy();
         },
     };
 }

src/lib/routes/client-api/metrics.test.ts

@@ -20,7 +20,6 @@ async function getSetup(opts?: IUnleashOptions) {
         destroy: () => {
             services.versionService.destroy();
             services.clientInstanceService.destroy();
-            services.apiTokenService.destroy();
         },
     };
 }

src/lib/routes/client-api/register.test.ts

@@ -17,7 +17,6 @@ async function getSetup() {
         destroy: () => {
             services.versionService.destroy();
             services.clientInstanceService.destroy();
-            services.apiTokenService.destroy();
         },
     };
 }

src/lib/routes/health-check.test.ts

@@ -18,7 +18,6 @@ async function getSetup() {
         destroy: () => {
             services.versionService.destroy();
             services.clientInstanceService.destroy();
-            services.apiTokenService.destroy();
         },
     };
 }

src/lib/services/api-token-service.ts

@@ -17,7 +17,6 @@ import {
 import { IApiTokenStore } from '../types/stores/api-token-store';
 import { FOREIGN_KEY_VIOLATION } from '../error/db-error';
 import BadDataError from '../error/bad-data-error';
-import { minutesToMilliseconds } from 'date-fns';
 import { IEnvironmentStore } from 'lib/types/stores/environment-store';
 import { constantTimeCompare } from '../util/constantTimeCompare';
 import {
@@ -50,10 +49,6 @@ export class ApiTokenService {
 
     private logger: Logger;
 
-    private timer: NodeJS.Timeout;
-
-    private seenTimer: NodeJS.Timeout;
-
     private activeTokens: IApiToken[] = [];
 
     private eventStore: IEventStore;
@@ -76,10 +71,6 @@ export class ApiTokenService {
         this.environmentStore = environmentStore;
         this.logger = config.getLogger('/services/api-token-service.ts');
         this.fetchActiveTokens();
-        this.timer = setInterval(
-            () => this.fetchActiveTokens(),
-            minutesToMilliseconds(1),
-        ).unref();
         this.updateLastSeen();
         if (config.authentication.initApiTokens.length > 0) {
             process.nextTick(async () =>
@@ -103,11 +94,6 @@ export class ApiTokenService {
             this.lastSeenSecrets = new Set<string>();
             await this.store.markSeenAt(toStore);
         }
-
-        this.seenTimer = setTimeout(
-            async () => this.updateLastSeen(),
-            minutesToMilliseconds(3),
-        ).unref();
     }
 
     public async getAllTokens(): Promise<IApiToken[]> {
@@ -286,11 +272,4 @@ export class ApiTokenService {
             return `${projects[0]}:${environment}.${randomStr}`;
         }
     }
-
-    destroy(): void {
-        clearInterval(this.timer);
-        clearTimeout(this.seenTimer);
-        this.timer = null;
-        this.seenTimer = null;
-    }
 }

src/lib/services/index.ts

@@ -40,6 +40,8 @@ import { InstanceStatsService } from './instance-stats-service';
 import { FavoritesService } from './favorites-service';
 import MaintenanceService from './maintenance-service';
 import ExportImportService from './export-import-service';
+import SchedulerService from './scheduler-service';
+import { minutesToMilliseconds } from 'date-fns';
 
 export const createServices = (
     stores: IUnleashStores,
@@ -139,6 +141,17 @@ export const createServices = (
         settingService,
     );
 
+    const schedulerService = new SchedulerService(config.getLogger);
+    schedulerService.schedule(
+        apiTokenService.fetchActiveTokens.bind(apiTokenService),
+        minutesToMilliseconds(1),
+    );
+
+    schedulerService.schedule(
+        apiTokenService.updateLastSeen.bind(apiTokenService),
+        minutesToMilliseconds(3),
+    );
+
     return {
         accessService,
         addonService,

src/lib/services/scheduler-service.test.ts

@@ -0,0 +1,95 @@
+import SchedulerService from './scheduler-service';
+
+function ms(timeMs) {
+    return new Promise((resolve) => setTimeout(resolve, timeMs));
+}
+
+const getLogger = () => {
+    const records = [];
+    const logger = () => ({
+        error(...args: any[]) {
+            records.push(args);
+        },
+        debug() {},
+        info() {},
+        warn() {},
+        fatal() {},
+        getRecords() {
+            return records;
+        },
+    });
+    logger.getRecords = () => records;
+
+    return logger;
+};
+
+test('Can schedule a single regular job', async () => {
+    const schedulerService = new SchedulerService(getLogger());
+    const job = jest.fn();
+
+    schedulerService.schedule(job, 10);
+    await ms(15);
+
+    expect(job).toBeCalledTimes(1);
+    schedulerService.stop();
+});
+
+test('Can schedule multiple jobs at the same interval', async () => {
+    const schedulerService = new SchedulerService(getLogger());
+    const job = jest.fn();
+    const anotherJob = jest.fn();
+
+    schedulerService.schedule(job, 10);
+    schedulerService.schedule(anotherJob, 10);
+    await ms(15);
+
+    expect(job).toBeCalledTimes(1);
+    expect(anotherJob).toBeCalledTimes(1);
+    schedulerService.stop();
+});
+
+test('Can schedule multiple jobs at the different intervals', async () => {
+    const schedulerService = new SchedulerService(getLogger());
+    const job = jest.fn();
+    const anotherJob = jest.fn();
+
+    schedulerService.schedule(job, 10);
+    schedulerService.schedule(anotherJob, 20);
+    await ms(25);
+
+    expect(job).toBeCalledTimes(2);
+    expect(anotherJob).toBeCalledTimes(1);
+    schedulerService.stop();
+});
+
+test('Can handle crash of a async job', async () => {
+    const logger = getLogger();
+    const schedulerService = new SchedulerService(logger);
+    const job = async () => {
+        await Promise.reject('async reason');
+    };
+
+    schedulerService.schedule(job, 10);
+    await ms(15);
+
+    schedulerService.stop();
+    expect(logger.getRecords()).toEqual([
+        ['scheduled job failed', 'async reason'],
+    ]);
+});
+
+test('Can handle crash of a sync job', async () => {
+    const logger = getLogger();
+    const schedulerService = new SchedulerService(logger);
+    const job = () => {
+        throw new Error('sync reason');
+    };
+
+    schedulerService.schedule(job, 10);
+    await ms(15);
+
+    schedulerService.stop();
+    expect(logger.getRecords()).toEqual([
+        ['scheduled job failed', new Error('sync reason')],
+    ]);
+});

src/lib/services/scheduler-service.ts

@@ -0,0 +1,27 @@
+import { Logger, LogProvider } from '../logger';
+
+export default class SchedulerService {
+    private intervalIds: NodeJS.Timer[] = [];
+
+    private logger: Logger;
+
+    constructor(getLogger: LogProvider) {
+        this.logger = getLogger('/services/scheduler-service.ts');
+    }
+
+    schedule(scheduledFunction: () => void, timeMs: number): void {
+        this.intervalIds.push(
+            setInterval(async () => {
+                try {
+                    await scheduledFunction();
+                } catch (e) {
+                    this.logger.error('scheduled job failed', e);
+                }
+            }, timeMs).unref(),
+        );
+    }
+
+    stop(): void {
+        this.intervalIds.forEach(clearInterval);
+    }
+}

src/test/e2e/helpers/test-helper.ts

@@ -46,7 +46,6 @@ async function createApp(
         services.versionService.destroy();
         services.clientInstanceService.destroy();
         services.clientMetricsServiceV2.destroy();
-        services.apiTokenService.destroy();
         services.proxyService.destroy();
     };