Fixing up the order of wait and close to ensure that queue promsies a…

…ctually resolve

src/deploy/functions/release.ts

@@ -5,7 +5,7 @@ import * as clc from "cli-color";
 
 import * as utils from "../../utils";
 import * as helper from "../../functionsDeployHelper";
-import { createDeploymentPlan } from "./deploymentPlanner";
+import { CloudFunctionTrigger, createDeploymentPlan } from "./deploymentPlanner";
 import * as tasks from "./tasks";
 import { getAppEngineLocation } from "../../functionsConfig";
 import { promptForFunctionDeletion } from "./prompts";
@@ -30,8 +30,8 @@ export async function release(context: any, options: any, payload: any) {
     context.existingFunctions,
     context.filters
   );
-  const cloudFunctionsQueue = new Queue<() => any, any>({});
-  const schedulerQueue = new Queue<() => any, any>({});
+  const cloudFunctionsQueue = new Queue<() => Promise<CloudFunctionTrigger|void>, void>({});
+  const schedulerQueue = new Queue<() => Promise<any>, void>({});
   const regionPromises = [];
 
   const retryFuncParams: tasks.RetryFunctionParams = {
@@ -88,9 +88,6 @@ export async function release(context: any, options: any, payload: any) {
         });
     }
   }
-
-  cloudFunctionsQueue.process();
-
   for (const fnName of fullDeployment.schedulesToDelete) {
     schedulerQueue
       .run(tasks.deleteScheduleTask(fnName, appEngineLocation))
@@ -101,21 +98,29 @@ export async function release(context: any, options: any, payload: any) {
         errorHandler.record("error", fnName, "delete schedule", err.message || "");
       });
   }
-  schedulerQueue.close();
+
+  // Once everything has been added to queues, starting processing.
+  const queuePromises = [cloudFunctionsQueue.wait(), schedulerQueue.wait()]; 
+  cloudFunctionsQueue.process();
   schedulerQueue.process();
+  schedulerQueue.close();
 
-  // Wait for the first function in each region to be deployed, and all the other calls to be queued,
-  // then close the queue.
-  await Promise.all(regionPromises);
+  // Wait until the second round of creates/updates are added to the queue before closing it.
+  await Promise.all(regionPromises)
   cloudFunctionsQueue.close();
 
+  // Wait for the first function in each region to be deployed, and all the other calls to be queued,
+  // then close the queue.
   // Wait for all of the deployments to complete.
-  await cloudFunctionsQueue.wait();
-  await schedulerQueue.wait();
+  await Promise.all(queuePromises);
+  // TODO: We should also await the scheduler queue. However, for reasons I don't understand,
+  // awaiting 2 queues makes it so none of the code below this executes.
+  // If I remove either queue, it works correctly.
+  // Not sure if this is a bug with queue or if I'm doing something subtly incorrect, but functions deploys are ~2 orders of magnitude longer
+  // than schedule deployments, and there are no deployments that contain only calls to scheduler, so this works for now.
+  // await schedulerQueue.wait();
   helper.logAndTrackDeployStats(cloudFunctionsQueue);
-  await helper.printTriggerUrls(projectId, sourceUrl);
   errorHandler.printWarnings();
   errorHandler.printErrors();
-  console.log("Deployment Complete!");
-  return;
+  return helper.printTriggerUrls(projectId, sourceUrl);
 }

src/deploy/functions/tasks.ts

@@ -14,7 +14,6 @@ import { getHumanFriendlyRuntimeName } from "../../parseRuntimeAndValidateSDK";
 import { deleteTopic } from "../../gcp/pubsub";
 import { DeploymentTimer } from "./deploymentTimer";
 import { ErrorHandler } from "./errorHandler";
-import { FirebaseError } from "../../error";
 
 // TODO: Tune this for better performance.
 const defaultPollerOptions = {
@@ -36,7 +35,7 @@ export function createFunctionTask(
   params: RetryFunctionParams,
   fn: CloudFunctionTrigger,
   onPoll?: (op: any) => any
-) {
+): () => Promise<CloudFunctionTrigger> {
   return async () => {
     utils.logBullet(
       clc.bold.cyan("functions: ") +
@@ -75,7 +74,7 @@ export function createFunctionTask(
       },
       defaultPollerOptions
     );
-    const operationResult = await pollOperation(pollerOptions);
+    const operationResult = await pollOperation<CloudFunctionTrigger>(pollerOptions);
     if (eventType === "https") {
       try {
         await cloudfunctions.setIamPolicy({
@@ -97,7 +96,7 @@ export function updateFunctionTask(
   params: RetryFunctionParams,
   fn: CloudFunctionTrigger,
   onPoll?: (op: any) => any
-) {
+): () => Promise<CloudFunctionTrigger> {
   return async () => {
     utils.logBullet(
       clc.bold.cyan("functions: ") +
@@ -136,7 +135,7 @@ export function updateFunctionTask(
       },
       defaultPollerOptions
     );
-    const operationResult = await pollOperation(pollerOptions);
+    const operationResult = await pollOperation<CloudFunctionTrigger>(pollerOptions);
     params.timer.endTimer(fn.name);
     return operationResult;
   };
@@ -161,7 +160,7 @@ export function deleteFunctionTask(params: RetryFunctionParams, fnName: string)
       },
       defaultPollerOptions
     );
-    const operationResult = await pollOperation(pollerOptions);
+    const operationResult = await pollOperation<void>(pollerOptions);
     params.timer.endTimer(fnName);
     return operationResult;
   };
@@ -171,14 +170,14 @@ export function upsertScheduleTask(
   params: RetryFunctionParams,
   fn: CloudFunctionTrigger,
   appEngineLocation: string
-) {
+): () => Promise<any> {
   return async () => {
     const job = helper.toJob(fn, appEngineLocation, params.projectId);
     return cloudscheduler.createOrReplaceJob(job);
   };
 }
 
-export function deleteScheduleTask(fnName: string, appEngineLocation: string) {
+export function deleteScheduleTask(fnName: string, appEngineLocation: string): () => Promise<void> {
   return async () => {
     const jobName = helper.getScheduleName(fnName, appEngineLocation);
     const topicName = helper.getTopicName(fnName);
@@ -212,8 +211,8 @@ export function deleteScheduleTask(fnName: string, appEngineLocation: string) {
 export function runRegionalFunctionDeployment(
   params: RetryFunctionParams,
   regionalDeployment: RegionalDeployment,
-  queue: Queue<any, any>
-): Promise<any> {
+  queue: Queue<() => Promise<any>, void>
+): Promise<void> {
   // Build an onPoll function to check for sourceToken and queue up the rest of the deployment.
   const onPollFn = (op: any) => {
     // We should run the rest of the regional deployment if we either:
@@ -261,8 +260,8 @@ export function runRegionalFunctionDeployment(
 function finishRegionalFunctionDeployment(
   params: RetryFunctionParams,
   regionalDeployment: RegionalDeployment,
-  queue: Queue<any, any>
-) {
+  queue: Queue<()=>Promise<any>, void>
+): void {
   for (const fn of regionalDeployment.functionsToCreate) {
     queue
       .run(createFunctionTask(params, fn))