fix: continueWaitTimeout is ignored: expose single centralized contin…

…ueWaitTimeout option (#2120) * fix: continueWaitTimeout is ignored: expose single centralized continueWaitTimeout option Fixes #225 * chore: Fix intermittent test fail
cube-js · Feb 17, 2021 · 2735d2c · 2735d2c
1 parent 0d97357
commit 2735d2c
Show file tree

Hide file tree

Showing 7 changed files with 20 additions and 5 deletions.
diff --git a/packages/cubejs-backend-shared/test/promises.test.ts b/packages/cubejs-backend-shared/test/promises.test.ts
@@ -338,7 +338,7 @@ describe('asyncMemoize', () => {
 
     expect(called).toEqual(2);
 
-    await pausePromise(500);
+    await pausePromise(800);
 
     expect(await memCall('test') !== firstCallRandomValue).toEqual(true);
 

diff --git a/packages/cubejs-query-orchestrator/src/orchestrator/PreAggregations.ts b/packages/cubejs-query-orchestrator/src/orchestrator/PreAggregations.ts
@@ -773,7 +773,8 @@ type PreAggregationsOptions = {
   loadCacheQueueOptions?: any;
   queueOptions?: object | ((dataSource: String) => object);
   redisPool?: any;
-  cacheAndQueueDriver?: 'redis' | 'memory'
+  continueWaitTimeout?: number;
+  cacheAndQueueDriver?: 'redis' | 'memory';
 };
 
 export class PreAggregations {
@@ -895,6 +896,8 @@ export class PreAggregations {
         logger: this.logger,
         cacheAndQueueDriver: this.options.cacheAndQueueDriver,
         redisPool: this.options.redisPool,
+        // Centralized continueWaitTimeout that can be overridden in queueOptions
+        continueWaitTimeout: this.options.continueWaitTimeout,
         ...(typeof this.options.queueOptions === 'function' ?
           this.options.queueOptions(dataSource) :
           this.options.queueOptions

diff --git a/packages/cubejs-query-orchestrator/src/orchestrator/QueryCache.ts b/packages/cubejs-query-orchestrator/src/orchestrator/QueryCache.ts
@@ -31,6 +31,7 @@ export class QueryCache {
       backgroundRenew?: Boolean;
       queueOptions?: object | ((dataSource: String) => object);
       redisPool?: any;
+      continueWaitTimeout?: number;
       cacheAndQueueDriver?: 'redis' | 'memory';
     } = {}
   ) {
@@ -202,6 +203,8 @@ export class QueryCache {
           logger: this.logger,
           cacheAndQueueDriver: this.options.cacheAndQueueDriver,
           redisPool: this.options.redisPool,
+          // Centralized continueWaitTimeout that can be overridden in queueOptions
+          continueWaitTimeout: this.options.continueWaitTimeout,
           ...(typeof this.options.queueOptions === 'function' ?
             this.options.queueOptions(dataSource) :
             this.options.queueOptions
@@ -227,6 +230,8 @@ export class QueryCache {
           logger: this.logger,
           cacheAndQueueDriver: this.options.cacheAndQueueDriver,
           redisPool: this.options.redisPool,
+          // Centralized continueWaitTimeout that can be overridden in queueOptions
+          continueWaitTimeout: this.options.continueWaitTimeout,
           ...this.options.externalQueueOptions
         }
       );

diff --git a/packages/cubejs-query-orchestrator/src/orchestrator/QueryOrchestrator.ts b/packages/cubejs-query-orchestrator/src/orchestrator/QueryOrchestrator.ts
@@ -13,6 +13,7 @@ interface QueryOrchestratorOptions {
   queryCacheOptions?: any;
   preAggregationsOptions?: any;
   rollupOnlyMode?: boolean;
+  continueWaitTimeout?: number;
 }
 
 export class QueryOrchestrator {
@@ -45,7 +46,7 @@ export class QueryOrchestrator {
     }
 
     const redisPool = cacheAndQueueDriver === 'redis' ? new RedisPool(options.redisPoolOptions) : undefined;
-    const { externalDriverFactory } = options;
+    const { externalDriverFactory, continueWaitTimeout } = options;
 
     this.driverFactory = driverFactory;
 
@@ -54,6 +55,7 @@ export class QueryOrchestrator {
         externalDriverFactory,
         cacheAndQueueDriver,
         redisPool,
+        continueWaitTimeout,
         ...options.queryCacheOptions,
       }
     );
@@ -63,6 +65,7 @@ export class QueryOrchestrator {
         externalDriverFactory,
         cacheAndQueueDriver,
         redisPool,
+        continueWaitTimeout,
         ...options.preAggregationsOptions
       }
     );

diff --git a/packages/cubejs-server-core/src/core/OrchestratorApi.ts b/packages/cubejs-server-core/src/core/OrchestratorApi.ts
@@ -8,10 +8,12 @@ export class OrchestratorApi {
   protected readonly orchestrator: QueryOrchestrator;
 
   protected readonly externalDriverFactory: any;
+
+  protected readonly continueWaitTimeout: number;
 
   public constructor(protected driverFactory, protected logger, protected options: any = {}) {
     const { externalDriverFactory } = options;
-
+    this.continueWaitTimeout = this.options.continueWaitTimeout || 5;
     this.orchestrator = new QueryOrchestrator(options.redisPrefix || 'STANDALONE', driverFactory, logger, options);
     this.driverFactory = driverFactory;
     this.externalDriverFactory = externalDriverFactory;
@@ -33,7 +35,7 @@ export class OrchestratorApi {
         this.orchestrator.loadRefreshKeys(query) :
         this.orchestrator.fetchQuery(query);
 
-      fetchQueryPromise = pt.timeout(fetchQueryPromise, 5 * 1000);
+      fetchQueryPromise = pt.timeout(fetchQueryPromise, this.continueWaitTimeout * 1000);
 
       const data = await fetchQueryPromise;
 

diff --git a/packages/cubejs-server-core/src/core/optionsValidate.ts b/packages/cubejs-server-core/src/core/optionsValidate.ts
@@ -70,6 +70,7 @@ const schemaOptions = Joi.object().keys({
     Joi.func(),
     Joi.object().keys({
       redisPrefix: Joi.string().allow(''),
+      continueWaitTimeout: Joi.number().min(0).integer(),
       queryCacheOptions: Joi.object().keys({
         refreshKeyRenewalThreshold: Joi.number().min(0).integer(),
         backgroundRenew: Joi.boolean(),

diff --git a/packages/cubejs-server-core/test/unit/index.test.ts b/packages/cubejs-server-core/test/unit/index.test.ts
@@ -100,6 +100,7 @@ describe('index.test', () => {
       allowUngroupedWithoutPrimaryKey: true,
       scheduledRefreshConcurrency: 4,
       orchestratorOptions: {
+        continueWaitTimeout: 10,
         redisPrefix: 'some-prefix',
         queryCacheOptions: {
           refreshKeyRenewalThreshold: 1000,