refactor(graph): make sure all tasks are included in process results

garden-service/src/commands/publish.ts

@@ -80,7 +80,7 @@ export async function publishModules(
   forceBuild: boolean,
   allowDirty: boolean,
 ): Promise<TaskResults> {
-  for (const module of modules) {
+  const tasks = modules.map(module => {
     const version = module.version
 
     if (version.dirtyTimestamp && !allowDirty) {
@@ -91,9 +91,8 @@ export async function publishModules(
       )
     }
 
-    const task = new PublishTask({ garden, log, module, forceBuild })
-    await garden.addTask(task)
-  }
+    return new PublishTask({ garden, log, module, forceBuild })
+  })
 
-  return await garden.processTasks()
+  return await garden.processTasks(tasks)
 }

garden-service/src/commands/run/module.ts

@@ -91,8 +91,7 @@ export class RunModuleCommand extends Command<Args, Opts> {
     await garden.actions.prepareEnvironment({ log })
 
     const pushTask = new PushTask({ garden, log, module, force: opts["force-build"] })
-    await garden.addTask(pushTask)
-    await garden.processTasks()
+    await garden.processTasks([pushTask])
 
     const command = args.command || []
 

garden-service/src/commands/run/service.ts

@@ -68,8 +68,7 @@ export class RunServiceCommand extends Command<Args, Opts> {
     await garden.actions.prepareEnvironment({ log })
 
     const pushTask = new PushTask({ garden, log, module, force: opts["force-build"] })
-    await garden.addTask(pushTask)
-    await garden.processTasks()
+    await garden.processTasks([pushTask])
 
     const dependencies = await graph.getServices(module.serviceDependencyNames)
     const runtimeContext = await prepareRuntimeContext(garden, graph, module, dependencies)

garden-service/src/commands/run/task.ts

@@ -60,9 +60,7 @@ export class RunTaskCommand extends Command<Args, Opts> {
     await garden.actions.prepareEnvironment({ log })
 
     const taskTask = await TaskTask.factory({ garden, graph, task, log, force: true, forceBuild: opts["force-build"] })
-    await garden.addTask(taskTask)
-
-    const result = (await garden.processTasks())[taskTask.getBaseKey()]
+    const result = (await garden.processTasks([taskTask]))[taskTask.getBaseKey()]
 
     if (!result.error) {
       log.info("")

garden-service/src/commands/run/test.ts

@@ -93,8 +93,7 @@ export class RunTestCommand extends Command<Args, Opts> {
     await garden.actions.prepareEnvironment({ log })
 
     const pushTask = new PushTask({ garden, log, module, force: opts["force-build"] })
-    await garden.addTask(pushTask)
-    await garden.processTasks()
+    await garden.processTasks([pushTask])
 
     const interactive = opts.interactive
     const deps = await graph.getDependencies("test", testConfig.name, false)

garden-service/src/garden.ts

@@ -299,12 +299,8 @@ export class Garden {
     return this.buildDir.clear()
   }
 
-  async addTask(task: BaseTask) {
-    await this.taskGraph.addTask(task)
-  }
-
-  async processTasks(): Promise<TaskResults> {
-    return this.taskGraph.processTasks()
+  async processTasks(tasks: BaseTask[]): Promise<TaskResults> {
+    return this.taskGraph.process(tasks)
   }
 
   /**

garden-service/src/process.ts

@@ -8,7 +8,7 @@
 
 import Bluebird = require("bluebird")
 import chalk from "chalk"
-import { padEnd, keyBy } from "lodash"
+import { padEnd, keyBy, flatten } from "lodash"
 
 import { Module } from "./types/module"
 import { Service } from "./types/service"
@@ -85,10 +85,7 @@ export async function processModules(
     log.info(divider)
   }
 
-  for (const module of modules) {
-    const tasks = await handler(graph, module)
-    await Bluebird.map(tasks, t => garden.addTask(t))
-  }
+  const tasks: BaseTask[] = flatten(await Bluebird.map(modules, module => handler(graph, module)))
 
   if (watch && !!logFooter) {
     garden.events.on("taskGraphProcessing", () => {
@@ -100,7 +97,7 @@ export async function processModules(
     })
   }
 
-  const results = await garden.processTasks()
+  const results = await garden.processTasks(tasks)
 
   if (!watch) {
     return {
@@ -164,11 +161,11 @@ export async function processModules(
       // Make sure the modules' versions are up to date.
       const changedModules = await graph.getModules(changedModuleNames)
 
-      await Bluebird.map(changedModules, async (m) => {
+      const moduleTasks = flatten(await Bluebird.map(changedModules, async (m) => {
         modulesByName[m.name] = m
-        return Bluebird.map(changeHandler!(graph, m), (task) => garden.addTask(task))
-      })
-      await garden.processTasks()
+        return changeHandler!(graph, m)
+      }))
+      await garden.processTasks(moduleTasks)
     })
   })
 

garden-service/src/task-graph.ts

@@ -65,18 +65,14 @@ export class TaskGraph {
     this.logEntryMap = {}
   }
 
-  async addTask(task: BaseTask): Promise<void> {
-    return this.opQueue.add(() => this.addTaskInternal(task))
-  }
-
-  async processTasks(): Promise<TaskResults> {
-    return this.opQueue.add(() => this.processTasksInternal())
+  async process(tasks: BaseTask[]): Promise<TaskResults> {
+    return this.opQueue.add(() => this.processTasksInternal(tasks))
   }
 
   /**
    * Rebuilds the dependency relationships between the TaskNodes in this.index, and updates this.roots accordingly.
    */
-  private async rebuild() {
+  private rebuild() {
     const taskNodes = this.index.getNodes()
 
     // this.taskDependencyCache will already have been populated at this point (happens in addTaskInternal).
@@ -95,9 +91,9 @@ export class TaskGraph {
     this.roots.setNodes(newRootNodes)
   }
 
-  private async addTaskInternal(task: BaseTask) {
+  private async addTask(task: BaseTask) {
     await this.addNodeWithDependencies(task)
-    await this.rebuild()
+    this.rebuild()
     if (this.index.getNode(task)) {
       this.garden.events.emit("taskPending", {
         addedAt: new Date(),
@@ -131,7 +127,11 @@ export class TaskGraph {
   /**
    * Process the graph until it's complete.
    */
-  private async processTasksInternal(): Promise<TaskResults> {
+  private async processTasksInternal(tasks: BaseTask[]): Promise<TaskResults> {
+    for (const task of tasks) {
+      await this.addTask(task)
+    }
+
     this.log.silly("")
     this.log.silly("TaskGraph: this.index before processing")
     this.log.silly("---------------------------------------")
@@ -155,7 +155,7 @@ export class TaskGraph {
         .slice(0, _this.concurrency - this.inProgress.length)
 
       batch.forEach(n => this.inProgress.addNode(n))
-      await this.rebuild()
+      this.rebuild()
 
       this.initLogging()
 
@@ -185,13 +185,13 @@ export class TaskGraph {
             result.error = error
             this.garden.events.emit("taskError", result)
             this.logTaskError(node, error)
-            await this.cancelDependants(node)
+            this.cancelDependants(node)
           } finally {
             results[baseKey] = result
             this.resultCache.put(baseKey, task.version.versionString, result)
           }
         } finally {
-          await this.completeTask(node, !result.error)
+          this.completeTask(node, !result.error)
         }
 
         return loop()
@@ -200,7 +200,7 @@ export class TaskGraph {
 
     await loop()
 
-    await this.rebuild()
+    this.rebuild()
 
     return results
   }
@@ -225,14 +225,14 @@ export class TaskGraph {
     }
   }
 
-  private async completeTask(node: TaskNode, success: boolean) {
+  private completeTask(node: TaskNode, success: boolean) {
     if (node.getDependencies().length > 0) {
       throw new TaskGraphError(`Task ${node.getKey()} still has unprocessed dependencies`)
     }
 
     this.remove(node)
     this.logTaskComplete(node, success)
-    await this.rebuild()
+    this.rebuild()
   }
 
   private remove(node: TaskNode) {
@@ -241,12 +241,12 @@ export class TaskGraph {
   }
 
   // Recursively remove node's dependants, without removing node.
-  private async cancelDependants(node: TaskNode) {
+  private cancelDependants(node: TaskNode) {
     for (const dependant of this.getDependants(node)) {
       this.logTaskComplete(dependant, false)
       this.remove(dependant)
     }
-    await this.rebuild()
+    this.rebuild()
   }
 
   private getDependants(node: TaskNode): TaskNode[] {

garden-service/test/unit/src/build-dir.ts

@@ -1,4 +1,3 @@
-import * as Bluebird from "bluebird"
 const nodetree = require("nodetree")
 import { join } from "path"
 import { pathExists, readdir } from "fs-extra"
@@ -69,17 +68,14 @@ describe("BuildDir", () => {
       await garden.clearBuilds()
       const graph = await garden.getConfigGraph()
       const modules = await graph.getModules()
-
-      await Bluebird.map(modules, async (module) => {
-        return garden.addTask(new BuildTask({
-          garden,
-          log,
-          module,
-          force: true,
-        }))
-      })
-
-      await garden.processTasks()
+      const tasks = modules.map(module => new BuildTask({
+        garden,
+        log,
+        module,
+        force: true,
+      }))
+
+      await garden.processTasks(tasks)
 
       const buildDirD = await garden.buildDir.buildPath("module-d")
       const buildDirE = await garden.buildDir.buildPath("module-e")

garden-service/test/unit/src/plugins/kubernetes/helm/common.ts

@@ -43,10 +43,8 @@ describe("Helm common functions", () => {
 
   async function buildModules() {
     const modules = await graph.getModules()
-    for (const module of modules) {
-      await garden.addTask(new BuildTask({ garden, log, module, force: false }))
-    }
-    const results = await garden.processTasks()
+    const tasks = modules.map(module => new BuildTask({ garden, log, module, force: false }))
+    const results = await garden.processTasks(tasks)
 
     const err = first(Object.values(results).map(r => r.error))