fix(query-orchestrator): QueryQueue - update heartbeat by queue id (#9817)

Cube Store allows updating the heartbeat by path or queue id. Using queue id provides better performance and fixes possible concurrency issues, such as slippage

Author: ovr

packages/cubejs-base-driver/src/queue-driver.interface.ts

@@ -1,8 +1,8 @@
 export type QueryDef = unknown;
 // Primary key of Queue item
 export type QueueId = string | number | bigint;
-// This was used as lock for Redis, deprecated.
-export type ProcessingId = string | number;
+// This was used as a lock for Redis, deprecated.
+export type ProcessingId = string | number | bigint;
 export type QueryKey = (string | [string, any[]]) & {
   persistent?: true,
 };
@@ -64,25 +64,25 @@ export interface QueueDriverConnectionInterface {
    * @param keyScore Redis specific thing
    * @param queryKey
    * @param orphanedTime
-   * @param queryHandler Our queue allow to use different handlers. For example query, cvsQuery, etc.
+   * @param queryHandler Our queue allows using different handlers. For example, query, cvsQuery, etc.
    * @param query
    * @param priority
    * @param options
    */
   addToQueue(keyScore: number, queryKey: QueryKey, orphanedTime: number, queryHandler: string, query: AddToQueueQuery, priority: number, options: AddToQueueOptions): Promise<AddToQueueResponse>;
-  // Return query keys which was sorted by priority and time
+  // Return query keys that were sorted by priority and time
   getToProcessQueries(): Promise<QueryKeysTuple[]>;
   getActiveQueries(): Promise<QueryKeysTuple[]>;
   getQueryDef(hash: QueryKeyHash, queueId: QueueId | null): Promise<QueryDef | null>;
-  // Queries which was added to queue, but was not processed and not needed
+  // Queries that were added to queue, but was not processed and not needed
   getOrphanedQueries(): Promise<QueryKeysTuple[]>;
-  // Queries which was not completed with old heartbeat
+  // Queries that were not completed with old heartbeat
   getStalledQueries(): Promise<QueryKeysTuple[]>;
   getQueryStageState(onlyKeys: boolean): Promise<QueryStageStateResponse>;
   updateHeartBeat(hash: QueryKeyHash, queueId: QueueId | null): Promise<void>;
   getNextProcessingId(): Promise<ProcessingId>;
   // Trying to acquire a lock for processing a queue item, this method can return null when
-  // multiple nodes tries to process the same query
+  // multiple nodes try to process the same query
   retrieveForProcessing(hash: QueryKeyHash, processingId: ProcessingId): Promise<RetrieveForProcessingResponse>;
   freeProcessingLock(hash: QueryKeyHash, processingId: ProcessingId, activated: unknown): Promise<void>;
   optimisticQueryUpdate(hash: QueryKeyHash, toUpdate: unknown, processingId: ProcessingId, queueId: QueueId | null): Promise<boolean>;

packages/cubejs-query-orchestrator/src/orchestrator/QueryQueue.js

@@ -807,7 +807,7 @@ export class QueryQueue {
               queryProcessHeartbeat = Date.now();
             }
 
-            return queueConnection.updateHeartBeat(queryKeyHashed);
+            return queueConnection.updateHeartBeat(queryKeyHashed, queueId);
           },
           this.heartBeatInterval * 1000
         );