Merge pull request #552 from brizental/shutdown-bug

Bug 1707891 - Do not clear deletion-request pings when clearing pings queue

Author: Beatriz Rizental

CHANGELOG.md

@@ -6,7 +6,7 @@
 
 [Full changelog](https://github.com/mozilla/glean.js/compare/v0.17.0...v0.18.0)
 
-* [#542](https://github.com/mozilla/glean.js/pull/542): Implement `shutdown` API.
+* [#542](https://github.com/mozilla/glean.js/pull/542), [#552](https://github.com/mozilla/glean.js/pull/552): Implement `shutdown` API.
 
 # v0.17.0 (2021-07-16)
 

glean/src/core/dispatcher.ts

@@ -27,18 +27,27 @@ export const enum DispatcherState {
 
 // The possible commands to be processed by the dispatcher.
 const enum Commands {
-  // The dispatcher must enqueue a new task.
+  // The dispatcher will enqueue a new task.
   //
   // This command is always followed by a concrete task for the dispatcher to execute.
   Task,
+  // Same as the `Task` command,
+  // but the task enqueued by this command is not cleared by the `Clear` command.
+  //
+  // # Note
+  //
+  // `Shutdown` will still clear these tasks.
+  //
+  // Unless unavoidable, prefer using the normal `Task`.
+  PersistentTask,
   // The dispatcher should stop executing the queued tasks.
   Stop,
   // The dispatcher should stop executing the queued tasks and clear the queue.
   Clear,
   // The dispatcher will clear the queue and go into the Shutdown state.
   Shutdown,
   // Exactly like a normal Task, but spawned for tests.
-  TestTask,
+  TestTask = "TestTask",
 }
 
 // A task the dispatcher knows how to execute.
@@ -47,13 +56,13 @@ type Task = () => Promise<void>;
 // An executable command.
 type Command = {
   task: Task,
-  command: Commands.Task,
+  command: Commands.Task | Commands.PersistentTask,
 } | {
   resolver: (value: void | PromiseLike<void>) => void,
   task: Task,
   command: Commands.TestTask,
 } | {
-  command: Exclude<Commands, Commands.Task | Commands.TestTask>,
+  command: Exclude<Commands, Commands.Task | Commands.TestTask | Commands.PersistentTask>,
 };
 
 /**
@@ -98,6 +107,19 @@ class Dispatcher {
     }
   }
 
+  /**
+   * Resolve all test resolvers.
+   *
+   * Used before clearing the queue in on a `Shutdown` or `Clear` command.
+   */
+  private unblockTestResolvers(): void {
+    this.queue.forEach(c => {
+      if (c.command === Commands.TestTask) {
+        c.resolver();
+      }
+    });
+  }
+
   /**
    * Executes all the commands in the queue, from oldest to newest.
    */
@@ -109,27 +131,23 @@ class Dispatcher {
         this.state = DispatcherState.Stopped;
         return;
       case(Commands.Shutdown):
-      case(Commands.Clear):
-        // Unblock test resolvers before clearing the queue.
-        this.queue.forEach(c => {
-          if (c.command === Commands.TestTask) {
-            c.resolver();
-          }
-        });
-
+        this.unblockTestResolvers();
         this.queue = [];
-        if (nextCommand.command === Commands.Clear) {
-          this.state = DispatcherState.Stopped;
-        } else {
-          this.state = DispatcherState.Shutdown;
-        }
-
+        this.state = DispatcherState.Shutdown;
         return;
+      case(Commands.Clear):
+        this.unblockTestResolvers();
+        this.queue = this.queue.filter(c =>
+          [Commands.PersistentTask, Commands.Shutdown].includes(c.command)
+        );
+        nextCommand = this.getNextCommand();
+        continue;
       case (Commands.TestTask):
         await this.executeTask(nextCommand.task);
         nextCommand.resolver();
         nextCommand = this.getNextCommand();
         continue;
+      case(Commands.PersistentTask):
       case(Commands.Task):
         await this.executeTask(nextCommand.task);
         nextCommand = this.getNextCommand();
@@ -237,6 +255,19 @@ class Dispatcher {
     });
   }
 
+  /**
+   * Works exactly like {@link launch},
+   * but enqueues a persistent task which is not cleared by the Clear command.
+   *
+   * @param task The task to enqueue.
+   */
+  launchPersistent(task: Task): void {
+    this.launchInternal({
+      task,
+      command: Commands.PersistentTask
+    });
+  }
+
   /**
    * Flushes the tasks enqueued while the dispatcher was uninitialized.
    *
@@ -268,9 +299,8 @@ class Dispatcher {
   /**
    * Enqueues a Clear command at the front of the queue and triggers execution.
    *
-   * The Clear command will remove all other tasks from the queue
-   * and put the dispatcher in a Stopped state after the command is executed.
-   * In order to re-start the dispatcher, call the `resume` method.
+   * The Clear command will remove all other tasks
+   * except for persistent tasks or shutdown tasks.
    *
    * # Note
    *
@@ -311,7 +341,7 @@ class Dispatcher {
    * Shutsdown the dispatcher.
    *
    * 1. Executes all tasks launched prior to this one.
-   * 2. Clears the queue of any tasks launched after this one.
+   * 2. Clears the queue of any tasks launched after this one (even persistent tasks).
    * 3. Puts the dispatcher in the `Shutdown` state.
    *
    * # Note
@@ -356,9 +386,10 @@ class Dispatcher {
       return;
     }
 
+    // Clear queue.
     this.clear();
-    // We need to wait for the clear command to be executed.
-    await this.testBlockOnQueue();
+    // Wait for the clear command and any persistent tasks that may still be in the queue.
+    await this.shutdown();
     this.state = DispatcherState.Uninitialized;
   }
 

glean/src/core/glean.ts

@@ -114,8 +114,10 @@ class Glean {
    * This function is only supposed to be called when telemetry is disabled.
    */
   private static async clearMetrics(): Promise<void> {
-    // Stop ongoing upload jobs and clear pending pings queue.
-    Glean.pingUploader.clearPendingPingsQueue();
+    // Clear enqueued upload jobs and clear pending pings queue.
+    //
+    // The only job that will still be sent is the deletion-request ping.
+    await Glean.pingUploader.clearPendingPingsQueue();
 
     // There is only one metric that we want to survive after clearing all
     // metrics: first_run_date. Here, we store its value
@@ -216,7 +218,11 @@ class Glean {
     Context.pingsDatabase = new PingsDatabase(Glean.platform.Storage);
     Context.errorManager = new ErrorManager();
 
-    Glean.instance._pingUploader = new PingUploader(correctConfig, Glean.platform, Context.pingsDatabase);
+    Glean.instance._pingUploader = new PingUploader(
+      correctConfig,
+      Glean.platform,
+      Context.pingsDatabase
+    );
 
     Context.pingsDatabase.attachObserver(Glean.pingUploader);
 
@@ -504,17 +510,14 @@ class Glean {
     // Get back to an uninitialized state.
     await Context.testUninitialize();
 
+    // Shutdown the current uploader.
+    //
+    // This is fine because a new uploader is created on initialize.
+    // It will also guarantee all pings to be sent before uninitializing.
+    await Glean.pingUploader?.shutdown();
+
     // Deregister all plugins
     testResetEvents();
-
-    // Await ongoing jobs and clear pending pings queue.
-    if (Glean.pingUploader) {
-      await Glean.pingUploader.testBlockOnPingsQueue();
-
-      // The first time tests run, before Glean is initialized, we are
-      // not guaranteed to have an uploader. Account for this.
-      Glean.pingUploader.clearPendingPingsQueue();
-    }
   }
 
   /**

glean/src/core/upload/index.ts

@@ -6,7 +6,7 @@ import { gzipSync, strToU8 } from "fflate";
 
 import type Platform from "../../platform/index.js";
 import type { Configuration } from "../config.js";
-import { GLEAN_VERSION } from "../constants.js";
+import { DELETION_REQUEST_PING_NAME, GLEAN_VERSION } from "../constants.js";
 import { Context } from "../context.js";
 import Dispatcher from "../dispatcher.js";
 import log, { LoggingLevel } from "../log.js";
@@ -51,6 +51,16 @@ interface QueuedPing extends PingInternalRepresentation {
   retries: number,
 }
 
+/**
+ * Whether or not a given queued ping is a deletion-request ping.
+ *
+ * @param ping The ping to verify.
+ * @returns Whether or not the ping is a deletion-request ping.
+ */
+function isDeletionRequest(ping: QueuedPing): boolean {
+  return ping.path.split("/")[3] === DELETION_REQUEST_PING_NAME;
+}
+
 // Error to be thrown in case the final ping body is larger than MAX_PING_BODY_SIZE.
 class PingBodyOverflowError extends Error {
   constructor(message?: string) {
@@ -116,8 +126,14 @@ class PingUploader implements PingsDatabaseObserver {
 
     // Add the ping to the list of pings being processsed.
     this.processing.push(ping);
+
+    // If the ping is a deletion-request ping, we want to enqueue it as a persistent task,
+    // so that clearing the queue does not clear it.
+    // eslint-disable-next-line @typescript-eslint/unbound-method
+    const launchFn = isDeletionRequest(ping) ? this.dispatcher.launchPersistent : this.dispatcher.launch;
+
     // Dispatch the uploading task.
-    this.dispatcher.launch(async (): Promise<void> => {
+    launchFn.bind(this.dispatcher)(async (): Promise<void> => {
       const status = await this.attemptPingUpload(ping);
       const shouldRetry = await this.processPingUploadResponse(ping.identifier, status);
 
@@ -221,7 +237,14 @@ class PingUploader implements PingsDatabaseObserver {
         finalPing.headers
       );
     } catch(e) {
-      log(LOG_TAG, ["Error trying to build ping request:", e], LoggingLevel.Warn);
+      log(
+        LOG_TAG,
+        [
+          "Error trying to build ping request:",
+          (e as Error).message
+        ],
+        LoggingLevel.Warn
+      );
       // An unrecoverable failure will make sure the offending ping is removed from the queue and
       // deleted from the database, which is what we want here.
       return {
@@ -321,9 +344,8 @@ class PingUploader implements PingsDatabaseObserver {
   }
 
   /**
-   * Shuts down internal dispatcher, after executing all previously enqueued ping requests.
-   *
-   * This is irreversible.
+   * Shuts down internal dispatcher,
+   * after executing all previously enqueued ping requests.
    *
    * @returns A promise that resolves once shutdown is complete.
    */
@@ -333,14 +355,25 @@ class PingUploader implements PingsDatabaseObserver {
 
   /**
    * Clears the pending pings queue.
+   *
+   * # Important
+   *
+   * This will _drop_ pending pings still enqueued.
+   * Only the `deletion-request` ping will still be processed.
    */
-  clearPendingPingsQueue(): void {
+  async clearPendingPingsQueue(): Promise<void> {
+    // Clears all tasks.
     this.dispatcher.clear();
-    this.processing = [];
-    // Create and initialize a new dispatcher so we don't need to wait
-    // on the previous one finishing execution.
+    // Wait for remaining jobs and shutdown.
     //
-    // It will only finish execution of the current task, all other queued tasks are dropped.
+    // The only jobs that may be left after clearing
+    // are `deletion-request` uploads.
+    await this.dispatcher.shutdown();
+    // At this poit we are sure the dispatcher queue is also empty,
+    // so we can empty the processing queue.
+    this.processing = [];
+
+    // Create and initialize a new dispatcher, since the `shutdown` state is irreversible.
     this.dispatcher = createAndInitializeDispatcher();
   }