[CORL-1002] Queue Improvements (#2931)

* fix: improved count handler for stories not found

* feat: removed performance-now

* feat: cleaned up processors, exposed counts

* feat: increased verb of schd jobs

* fix: removed dead code

src/core/server/queue/Task.ts

@@ -1,8 +1,10 @@
-import Queue, { Job, Queue as QueueType } from "bull";
+import Queue, { Job, JobCounts, Queue as QueueType } from "bull";
 import Logger from "bunyan";
 
 import TIME from "coral-common/time";
+import { createTimer } from "coral-server/helpers";
 import logger from "coral-server/logger";
+import { TenantResource } from "coral-server/models/tenant";
 
 export type JobProcessor<T, U = void> = (job: Job<T>) => Promise<U>;
 
@@ -13,10 +15,10 @@ export interface TaskOptions<T, U = void> {
   queue: Queue.QueueOptions;
 }
 
-export default class Task<T, U = any> {
-  private options: TaskOptions<T, U>;
-  private queue: QueueType<T>;
-  private log: Logger;
+export default class Task<T extends TenantResource, U = any> {
+  private readonly options: Required<TaskOptions<T, U>>;
+  private readonly queue: QueueType<T>;
+  private readonly log: Logger;
 
   constructor({
     jobName,
@@ -34,6 +36,9 @@ export default class Task<T, U = any> {
         // with completed entries if we don't need to.
         removeOnComplete: true,
 
+        // Remove the job if it fails after all attempts.
+        removeOnFail: true,
+
         // By default, configure jobs to use an exponential backoff strategy.
         backoff: {
           type: "exponential",
@@ -52,6 +57,10 @@ export default class Task<T, U = any> {
     // TODO: (wyattjoh) attach event handlers to the queue for metrics via: https://github.com/OptimalBits/bull/blob/develop/REFERENCE.md#events
   }
 
+  public async counts(): Promise<JobCounts> {
+    return this.queue.getJobCounts();
+  }
+
   /**
    * Add will add the job to the queue to get processed. It's not needed to
    * handle the job after it has been created.
@@ -62,7 +71,7 @@ export default class Task<T, U = any> {
     // Create the job.
     const job = await this.queue.add(data, this.options.jobOptions);
 
-    this.log.trace({ jobID: job.id }, "added job to queue");
+    this.log.info({ jobID: job.id }, "added job to queue");
     return job;
   }
 
@@ -77,15 +86,19 @@ export default class Task<T, U = any> {
         true
       );
 
-      log.trace("processing job from queue");
+      const timer = createTimer();
+      log.info("processing job from queue");
 
       try {
         // Send the job off to the job processor to be handled.
         const promise: U = await this.options.jobProcessor(job);
-        log.trace("processing completed");
+
+        // Log it!
+        log.info({ took: timer() }, "processing completed");
+
         return promise;
       } catch (err) {
-        log.error({ err }, "job failed to process");
+        log.error({ err, took: timer() }, "job failed to process");
         throw err;
       }
     });

src/core/server/queue/index.ts

@@ -13,9 +13,7 @@ import { createRejectorTask, RejectorQueue } from "./tasks/rejector";
 import { createScraperTask, ScraperQueue } from "./tasks/scraper";
 import { createWebhookTask, WebhookQueue } from "./tasks/webhook";
 
-const createQueueOptions = async (
-  config: Config
-): Promise<Queue.QueueOptions> => {
+const createQueueOptions = (config: Config): Queue.QueueOptions => {
   const client = createRedisClient(config);
   const subscriber = createRedisClient(config);
 
@@ -60,10 +58,13 @@ export interface TaskQueue {
   rejector: RejectorQueue;
 }
 
-export async function createQueue(options: QueueOptions): Promise<TaskQueue> {
+export function createQueue(options: QueueOptions): TaskQueue {
+  // Pull some options out.
+  const { config } = options;
+
   // Create the processor queue options. This holds references to the Redis
   // clients that are shared per queue.
-  const queueOptions = await createQueueOptions(options.config);
+  const queueOptions = createQueueOptions(config);
 
   // Attach process functions to the various tasks in the queue.
   const mailer = createMailerTask(queueOptions, options);
@@ -73,7 +74,6 @@ export async function createQueue(options: QueueOptions): Promise<TaskQueue> {
     ...options,
   });
   const webhook = createWebhookTask(queueOptions, options);
-
   const rejector = createRejectorTask(queueOptions, options);
 
   // Return the tasks + client.

src/core/server/queue/tasks/Task.ts

@@ -1,70 +0,0 @@
-import Queue, { Job, Queue as QueueType } from "bull";
-import logger from "coral-server/logger";
-
-export interface TaskOptions<T, U = any> {
-  jobName: string;
-  jobProcessor: (job: Job<T>) => Promise<U>;
-  queue: Queue.QueueOptions;
-}
-
-export default class Task<T, U = any> {
-  private options: TaskOptions<T, U>;
-  private queue: QueueType<T>;
-
-  constructor(options: TaskOptions<T, U>) {
-    this.queue = new Queue(options.jobName, options.queue);
-    this.options = options;
-
-    // Sets up and attaches the job processor to the queue.
-    this.setupAndAttachProcessor();
-  }
-
-  /**
-   * Add will add the job to the queue to get processed. It's not needed to
-   * handle the job after it has been created.
-   *
-   * @param data the data for the job to add.
-   */
-  public async add(data: T) {
-    const job = await this.queue.add(data, {
-      // We always remove the job when it's complete, no need to fill up Redis
-      // with completed entries if we don't need to.
-      removeOnComplete: true,
-    });
-
-    logger.trace(
-      { jobID: job.id, jobName: this.options.jobName },
-      "added job to queue"
-    );
-    return job;
-  }
-
-  private setupAndAttachProcessor() {
-    this.queue.process(async (job: Job<T>) => {
-      logger.trace(
-        { jobID: job.id, jobName: this.options.jobName },
-        "processing job from queue"
-      );
-
-      try {
-        // Send the job off to the job processor to be handled.
-        const promise: U = await this.options.jobProcessor(job);
-        logger.trace(
-          { jobID: job.id, jobName: this.options.jobName },
-          "processing completed"
-        );
-
-        return promise;
-      } catch (err) {
-        logger.error({ err }, "failed to process job from queue");
-
-        throw err;
-      }
-    });
-
-    logger.trace(
-      { jobName: this.options.jobName },
-      "registered processor for job type"
-    );
-  }
-}

src/core/server/queue/tasks/mailer/index.ts

@@ -1,6 +1,6 @@
 import Queue from "bull";
-import now from "performance-now";
 
+import { createTimer } from "coral-server/helpers";
 import logger from "coral-server/logger";
 import Task from "coral-server/queue/Task";
 import MailerContent from "coral-server/queue/tasks/mailer/content";
@@ -37,6 +37,10 @@ export class MailerQueue {
     this.tenantCache = options.tenantCache;
   }
 
+  public async counts() {
+    return this.task.counts();
+  }
+
   public async add({ template, tenantID, message: { to } }: MailerInput) {
     const log = logger.child(
       {
@@ -61,7 +65,7 @@ export class MailerQueue {
       return;
     }
 
-    const startTemplateGenerationTime = now();
+    const timer = createTimer();
 
     let html: string;
     try {
@@ -74,8 +78,7 @@ export class MailerQueue {
     }
 
     // Compute the end time.
-    const responseTime = Math.round(now() - startTemplateGenerationTime);
-    log.trace({ responseTime }, "finished template generation");
+    log.trace({ took: timer() }, "finished template generation");
 
     // Return the job that'll add the email to the queue to be processed later.
     return this.task.add({

src/core/server/queue/tasks/mailer/processor.ts

@@ -11,11 +11,11 @@ import { camelCase, isNil } from "lodash";
 import { Db } from "mongodb";
 import { createTransport } from "nodemailer";
 import { Options } from "nodemailer/lib/smtp-connection";
-import now from "performance-now";
 
 import { LanguageCode } from "coral-common/helpers/i18n/locales";
 import { Config } from "coral-server/config";
 import { InternalError } from "coral-server/errors";
+import { createTimer } from "coral-server/helpers";
 import logger from "coral-server/logger";
 import { Tenant } from "coral-server/models/tenant";
 import { I18n, translate } from "coral-server/services/i18n";
@@ -245,7 +245,7 @@ export const createJobProcessor = (options: MailProcessorOptions) => {
     // Construct the fromAddress.
     const fromAddress = fromName ? `${fromName} <${fromEmail}>` : fromEmail;
 
-    const startTemplateGenerationTime = now();
+    const templateGenerationTimer = createTimer();
 
     // Get the message to send.
     let message: Message;
@@ -261,9 +261,10 @@ export const createJobProcessor = (options: MailProcessorOptions) => {
       throw new InternalError(e, "could not translate the message");
     }
 
-    // Compute the end time.
-    const responseTime = Math.round(now() - startTemplateGenerationTime);
-    log.trace({ responseTime }, "finished mail translation");
+    log.trace(
+      { responseTime: templateGenerationTimer() },
+      "finished mail translation"
+    );
 
     let transport = cache.get(tenantID);
     if (!transport) {
@@ -300,7 +301,7 @@ export const createJobProcessor = (options: MailProcessorOptions) => {
 
     log.debug("starting to send the email");
 
-    const startMessageSendTime = now();
+    const messageSendTimer = createTimer();
 
     try {
       // Send the mail message.
@@ -309,8 +310,6 @@ export const createJobProcessor = (options: MailProcessorOptions) => {
       throw new InternalError(e, "could not send email");
     }
 
-    // Compute the end time.
-    const messageSendResponseTime = Math.round(now() - startMessageSendTime);
-    log.debug({ responseTime: messageSendResponseTime }, "sent the email");
+    log.debug({ responseTime: messageSendTimer() }, "sent the email");
   };
 };

src/core/server/queue/tasks/notifier/index.ts

@@ -22,46 +22,30 @@ interface Options {
   signingConfig: JWTSigningConfig;
 }
 
-/**
- * NotifierQueue is designed to handle creating and queuing notifications
- * that could be sent to users.
- */
-export class NotifierQueue {
-  private task: Task<NotifierData>;
-
-  constructor(queue: Queue.QueueOptions, options: Options) {
-    const registry = new Map<CoralEventType, NotificationCategory[]>();
-
-    // Notification categories have been grouped by their event name so that
-    // each event emitted need only access the associated notification once.
-    for (const category of categories) {
-      for (const event of category.events as CoralEventType[]) {
-        let handlers = registry.get(event);
-        if (!handlers) {
-          handlers = [];
-        }
-        handlers.push(category);
-        registry.set(event, handlers);
-      }
-    }
-
-    this.task = new Task({
-      jobName: JOB_NAME,
-      jobProcessor: createJobProcessor({ registry, ...options }),
-      queue,
-    });
-  }
-
-  public async add(data: NotifierData) {
-    return this.task.add(data);
-  }
-
-  public process() {
-    return this.task.process();
-  }
-}
+export type NotifierQueue = Task<NotifierData>;
 
 export const createNotifierTask = (
   queue: Queue.QueueOptions,
   options: Options
-) => new NotifierQueue(queue, options);
+) => {
+  const registry = new Map<CoralEventType, NotificationCategory[]>();
+
+  // Notification categories have been grouped by their event name so that
+  // each event emitted need only access the associated notification once.
+  for (const category of categories) {
+    for (const event of category.events as CoralEventType[]) {
+      let handlers = registry.get(event);
+      if (!handlers) {
+        handlers = [];
+      }
+      handlers.push(category);
+      registry.set(event, handlers);
+    }
+  }
+
+  return new Task({
+    jobName: JOB_NAME,
+    jobProcessor: createJobProcessor({ registry, ...options }),
+    queue,
+  });
+};

src/core/server/queue/tasks/notifier/processor.ts

@@ -79,51 +79,42 @@ export const createJobProcessor = ({
 
     log.debug("starting to handle a notify operation");
 
-    try {
-      // Get all the handlers that are active for this channel.
-      const categories = registry.get(input.type);
-      if (!categories || categories.length === 0) {
-        return;
-      }
-
-      // Grab the tenant from the cache.
-      const tenant = await tenantCache.retrieveByID(tenantID);
-      if (!tenant) {
-        throw new Error("tenant not found with ID");
-      }
-
-      // Create a notification context to handle processing notifications.
-      const ctx = new NotificationContext({
-        mongo,
-        config,
-        signingConfig,
-        tenant,
-        now,
-      });
-
-      // For each of the handler's we need to process, we should iterate to
-      // generate their notifications.
-      let notifications = await handleHandlers(ctx, categories, input);
-
-      // Check to see if some of the other notifications that are queued
-      // had this notification superseded.
-      notifications = notifications.filter(filterSuperseded);
-
-      // Send all the notifications now.
-      await processNewNotifications(
-        ctx,
-        notifications.map(({ notification }) => notification),
-        mailerQueue
-      );
-
-      log.debug(
-        { notifications: notifications.length },
-        "notifications handled"
-      );
-    } catch (err) {
-      log.error({ err }, "could not handle the notifications");
-
-      throw err;
+    // Get all the handlers that are active for this channel.
+    const categories = registry.get(input.type);
+    if (!categories || categories.length === 0) {
+      return;
     }
+
+    // Grab the tenant from the cache.
+    const tenant = await tenantCache.retrieveByID(tenantID);
+    if (!tenant) {
+      throw new Error("tenant not found with ID");
+    }
+
+    // Create a notification context to handle processing notifications.
+    const ctx = new NotificationContext({
+      mongo,
+      config,
+      signingConfig,
+      tenant,
+      now,
+    });
+
+    // For each of the handler's we need to process, we should iterate to
+    // generate their notifications.
+    let notifications = await handleHandlers(ctx, categories, input);
+
+    // Check to see if some of the other notifications that are queued
+    // had this notification superseded.
+    notifications = notifications.filter(filterSuperseded);
+
+    // Send all the notifications now.
+    await processNewNotifications(
+      ctx,
+      notifications.map(({ notification }) => notification),
+      mailerQueue
+    );
+
+    log.debug({ notifications: notifications.length }, "notifications handled");
   };
 };

src/core/server/queue/tasks/rejector.ts

@@ -1,7 +1,7 @@
 import Queue, { Job } from "bull";
 import { Db } from "mongodb";
-import now from "performance-now";
 
+import { createTimer } from "coral-server/helpers";
 import logger from "coral-server/logger";
 import {
   Comment,
@@ -61,49 +61,49 @@ const createJobProcessor = ({
     true
   );
   // Mark the start time.
-  const startTime = now();
+  const timer = createTimer();
+
   log.debug("starting to reject author comments");
+
   // Get the tenant.
   const tenant = await tenantCache.retrieveByID(tenantID);
   if (!tenant) {
     log.error("referenced tenant was not found");
     return;
   }
+
   // Get the current time.
   const currentTime = new Date();
-  try {
-    // Find all comments written by the author that should be rejected.
-    let connection = await getBatch(mongo, tenantID, authorID);
-    while (connection.nodes.length > 0) {
-      for (const comment of connection.nodes) {
-        // Get the latest revision of the comment.
-        const revision = getLatestRevision(comment);
-        // Reject the comment.
-        await rejectComment(
-          mongo,
-          redis,
-          null,
-          tenant,
-          comment.id,
-          revision.id,
-          moderatorID,
-          currentTime
-        );
-      }
-      // If there was not another page, abort processing.
-      if (!connection.pageInfo.hasNextPage) {
-        break;
-      }
-      // Load the next page.
-      connection = await getBatch(mongo, tenantID, authorID, connection);
+
+  // Find all comments written by the author that should be rejected.
+  let connection = await getBatch(mongo, tenantID, authorID);
+  while (connection.nodes.length > 0) {
+    for (const comment of connection.nodes) {
+      // Get the latest revision of the comment.
+      const revision = getLatestRevision(comment);
+
+      // Reject the comment.
+      await rejectComment(
+        mongo,
+        redis,
+        null,
+        tenant,
+        comment.id,
+        revision.id,
+        moderatorID,
+        currentTime
+      );
     }
-  } catch (err) {
-    log.error({ err }, "could not reject the author's comments");
-    throw err;
+    // If there was not another page, abort processing.
+    if (!connection.pageInfo.hasNextPage) {
+      break;
+    }
+    // Load the next page.
+    connection = await getBatch(mongo, tenantID, authorID, connection);
   }
+
   // Compute the end time.
-  const took = Math.round(now() - startTime);
-  log.debug({ took }, "rejected the author's comments");
+  log.debug({ took: timer() }, "rejected the author's comments");
 };
 
 export type RejectorQueue = Task<RejectorData>;

src/core/server/queue/tasks/scraper.ts

@@ -1,8 +1,8 @@
 import Queue, { Job } from "bull";
 import { Db } from "mongodb";
-import now from "performance-now";
 
 import { Config } from "coral-server/config";
+import { createTimer } from "coral-server/helpers";
 import logger from "coral-server/logger";
 import Task from "coral-server/queue/Task";
 import { scrape } from "coral-server/services/stories/scraper";
@@ -39,22 +39,13 @@ const createJobProcessor = ({
   );
 
   // Mark the start time.
-  const startTime = now();
+  const timer = createTimer();
 
   log.debug("starting to scrape the story");
-
-  try {
-    await scrape(mongo, config, tenantID, storyID, storyURL);
-  } catch (err) {
-    log.error({ err }, "could not scrape the story");
-
-    throw err;
-  }
+  await scrape(mongo, config, tenantID, storyID, storyURL);
 
   // Compute the end time.
-  const responseTime = Math.round(now() - startTime);
-
-  log.debug({ responseTime }, "scraped the story");
+  log.debug({ responseTime: timer() }, "scraped the story");
 };
 
 export type ScraperQueue = Task<ScraperData>;

src/core/server/queue/tasks/webhook/processor.ts

@@ -1,10 +1,10 @@
 import crypto from "crypto";
 import { Redis } from "ioredis";
 import { Db } from "mongodb";
-import getNow from "performance-now";
 
 import { Config } from "coral-server/config";
 import { CoralEventPayload } from "coral-server/events/event";
+import { createTimer } from "coral-server/helpers";
 import logger from "coral-server/logger";
 import {
   filterActiveSecrets,
@@ -170,17 +170,16 @@ export function createJobProcessor({
     );
 
     // Send the request.
-    const startedSendingAt = getNow();
+    const timer = createTimer();
     const res = await fetch(endpoint.url, options);
-    const took = getNow() - startedSendingAt;
     if (res.ok) {
       log.info(
-        { took, responseStatus: res.status },
+        { took: timer(), responseStatus: res.status },
         "finished sending webhook"
       );
     } else {
       log.warn(
-        { took, responseStatus: res.status },
+        { took: timer(), responseStatus: res.status },
         "failed to deliver webhook"
       );
     }