refactor(query-orchestrator): LocalQueueDriver - drop processingId

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

@@ -1,8 +1,6 @@
 export type QueryDef = any;
 // Primary key of Queue item
 export type QueueId = string | number | bigint;
-// This was used as a lock for Redis, deprecated.
-export type ProcessingId = string | number | bigint;
 export type QueryKey = (string | [string, any[]]) & {
   persistent?: true,
 };
@@ -12,25 +10,17 @@ export type QueryKeysTuple = [keyHash: QueryKeyHash, queueId: QueueId | null /**
 export type GetActiveAndToProcessResponse = [active: QueryKeysTuple[], toProcess: QueryKeysTuple[]];
 export type AddToQueueResponse = [added: number, queueId: QueueId | null, queueSize: number, addedToQueueTime: number];
 export type QueryStageStateResponse = [active: string[], toProcess: string[]] | [active: string[], toProcess: string[], defs: Record<string, QueryDef>];
-export type RetrieveForProcessingSuccess = [
-  added: unknown,
+/**
+ * `added` is `1` when the queue item was moved from pending to active by this call, `0` otherwise.
+ */
+export type RetrieveForProcessingResponse = [
+  added: number,
   // QueueId is required for Cube Store, other providers don't support it
   queueId: QueueId | null,
   active: QueryKeyHash[],
   pending: number,
-  def: QueryDef,
-  lockAquired: true
-];
-export type RetrieveForProcessingFail = [
-  added: unknown,
-  // QueueId is required for Cube Store, other providers don't support it
-  queueId: QueueId | null,
-  active: QueryKeyHash[],
-  pending: number,
-  def: null,
-  lockAquired: false
-];
-export type RetrieveForProcessingResponse = RetrieveForProcessingSuccess | RetrieveForProcessingFail | null;
+  def: QueryDef | null,
+] | null;
 
 export interface AddToQueueQuery {
   isJob: boolean,
@@ -64,15 +54,15 @@ export interface QueueDriverConnectionInterface {
    * Adds specified by the queryKey query to the queue, returns tuple
    * with the operation result.
    *
-   * @param keyScore Redis specific thing
    * @param queryKey
-   * @param orphanedTime
+   * @param orphanedTime Ignored by every current driver: the per item orphaned deadline is
+   *   derived from options.orphanedTimeout (in seconds) instead.
    * @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>;
+  addToQueue(queryKey: QueryKey, orphanedTime: number, queryHandler: string, query: AddToQueueQuery, priority: number, options: AddToQueueOptions): Promise<AddToQueueResponse>;
   // Return query keys which was sorted by priority and time
   getToProcessQueries(): Promise<QueryKeysTuple[]>;
   getActiveQueries(): Promise<QueryKeysTuple[]>;
@@ -83,17 +73,18 @@ export interface QueueDriverConnectionInterface {
   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
-  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>;
+  // Atomically moves a pending queue item to active, which is what stops multiple nodes
+  // from processing the same query. Returns `added: 0` when the item is missing, already
+  // active, or the queue is at its concurrency limit - in all of those cases nothing is
+  // mutated, so there is nothing for the caller to roll back.
+  retrieveForProcessing(hash: QueryKeyHash): Promise<RetrieveForProcessingResponse>;
+  optimisticQueryUpdate(hash: QueryKeyHash, toUpdate: unknown, queueId: QueueId | null): Promise<boolean>;
   cancelQuery(queryKey: QueryKey, queueId: QueueId | null): Promise<QueryDef | null>;
   getQueryAndRemove(hash: QueryKeyHash, queueId: QueueId | null): Promise<[QueryDef]>;
-  setResultAndRemoveQuery(hash: QueryKeyHash, executionResult: any, processingId: ProcessingId, queueId: QueueId | null): Promise<unknown>;
+  // Returns false when the queue item is gone (cancelled or orphaned while it was executing),
+  // which means the result was dropped.
+  setResultAndRemoveQuery(hash: QueryKeyHash, executionResult: any, queueId: QueueId | null): Promise<unknown>;
   release(): void;
-  //
   getQueriesToCancel(): Promise<QueryKeysTuple[]>
   // @deprecated
   getActiveAndToProcess(): Promise<GetActiveAndToProcessResponse>;

packages/cubejs-cubestore-driver/src/CubeStoreQueueDriver.ts

@@ -11,7 +11,6 @@ import {
   AddToQueueResponse,
   QueryKey,
   QueryKeyHash,
-  ProcessingId,
   QueueId,
   GetActiveAndToProcessResponse,
   QueryKeysTuple,
@@ -68,7 +67,6 @@ export class CubestoreQueueDriverConnection implements QueueDriverConnectionInte
   }
 
   public async addToQueue(
-    _keyScore: number,
     queryKey: QueryKey,
     _orphanedTime: number,
     queryHandler: string,
@@ -132,10 +130,6 @@ export class CubestoreQueueDriverConnection implements QueueDriverConnectionInte
     return null;
   }
 
-  public async freeProcessingLock(_hash: QueryKeyHash, _processingId: string, _activated: unknown): Promise<void> {
-    // nothing to do
-  }
-
   public async getActiveQueries(): Promise<QueryKeysTuple[]> {
     const rows = await this.driver.query<CubeStoreListResponse>('QUEUE ACTIVE ?', [
       this.options.redisQueuePrefix
@@ -185,17 +179,6 @@ export class CubestoreQueueDriverConnection implements QueueDriverConnectionInte
     ];
   }
 
-  public async getNextProcessingId(): Promise<number | string> {
-    const rows = await this.driver.query('CACHE INCR ?', [
-      `${this.options.redisQueuePrefix}:PROCESSING_COUNTER`
-    ]);
-    if (rows && rows.length) {
-      return rows[0].value;
-    }
-
-    throw new Error('Unable to get next processing id');
-  }
-
   public async getQueryStageState(onlyKeys: boolean): Promise<QueryStageStateResponse> {
     const rows = await this.driver.query<CubeStoreListResponse & { payload: string }>(`QUEUE LIST ${onlyKeys ? '?' : 'WITH_PAYLOAD ?'}`, [
       this.options.redisQueuePrefix
@@ -297,7 +280,7 @@ export class CubestoreQueueDriverConnection implements QueueDriverConnectionInte
     return null;
   }
 
-  public async optimisticQueryUpdate(hash: QueryKeyHash, toUpdate: unknown, _processingId: ProcessingId, queueId: QueueId): Promise<boolean> {
+  public async optimisticQueryUpdate(hash: QueryKeyHash, toUpdate: unknown, queueId: QueueId): Promise<boolean> {
     await this.driver.query('QUEUE MERGE_EXTRA ? ?', [
       // queryKeyHash as compatibility fallback
       queueId || this.prefixKey(hash),
@@ -311,7 +294,7 @@ export class CubestoreQueueDriverConnection implements QueueDriverConnectionInte
     // nothing to release
   }
 
-  public async retrieveForProcessing(hash: QueryKeyHash, _processingId: string): Promise<RetrieveForProcessingResponse> {
+  public async retrieveForProcessing(hash: QueryKeyHash): Promise<RetrieveForProcessingResponse> {
     const rows = await this.driver.query<{ id: string /* cube store convert int64 to string */, active: string | null, pending: string, payload: string, extra: string | null }>('QUEUE RETRIEVE EXTENDED CONCURRENCY ? ?', [
       this.options.concurrency,
       this.prefixKey(hash),
@@ -329,15 +312,17 @@ export class CubestoreQueueDriverConnection implements QueueDriverConnectionInte
           active,
           pending,
           def,
-          true
         ];
       } else {
+        // NotEnoughConcurrency, NotFound, LockFailed or ExclusiveAccessFailed. Cube Store
+        // returns all of them as an empty EXTENDED row and mutates nothing.
         return [
-          0, null, active, pending, null, false
+          0, null, active, pending, null
         ];
       }
     }
 
+    // Old Cube Store without EXTENDED support returns no rows at all on failure
     return null;
   }
 
@@ -354,7 +339,7 @@ export class CubestoreQueueDriverConnection implements QueueDriverConnectionInte
     return null;
   }
 
-  public async setResultAndRemoveQuery(hash: QueryKeyHash, executionResult: unknown, _processingId: ProcessingId, queueId: QueueId): Promise<boolean> {
+  public async setResultAndRemoveQuery(hash: QueryKeyHash, executionResult: unknown, queueId: QueueId): Promise<boolean> {
     const rows = await this.driver.query('QUEUE ACK ? ?', [
       // queryKeyHash as compatibility fallback
       queueId || this.prefixKey(hash),