diff --git a/taskcluster/taskgraph/generator.py b/taskcluster/taskgraph/generator.py
index 0e3c9e512a30..00586e4bcccf 100644
--- a/taskcluster/taskgraph/generator.py
+++ b/taskcluster/taskgraph/generator.py
@@ -321,8 +321,8 @@ class TaskGraphGenerator(object):
             always_target_tasks = set()
         logger.info('Adding %d tasks with `always_target` attribute' % (
                     len(always_target_tasks) - len(always_target_tasks & target_tasks)))
-        target_graph = full_task_graph.graph.transitive_closure(
-            target_tasks | docker_image_tasks | always_target_tasks)
+        requested_tasks = target_tasks | docker_image_tasks | always_target_tasks
+        target_graph = full_task_graph.graph.transitive_closure(requested_tasks)
         target_task_graph = TaskGraph(
             {l: all_tasks[l] for l in target_graph.nodes},
             target_graph)
@@ -342,6 +342,7 @@ class TaskGraphGenerator(object):
 
         optimized_task_graph, label_to_taskid = optimize_task_graph(
             target_task_graph,
+            requested_tasks,
             parameters,
             do_not_optimize,
             self._decision_task_id,
diff --git a/taskcluster/taskgraph/optimize/__init__.py b/taskcluster/taskgraph/optimize/__init__.py
index 817d02019a3d..631d8761c33b 100644
--- a/taskcluster/taskgraph/optimize/__init__.py
+++ b/taskcluster/taskgraph/optimize/__init__.py
@@ -39,7 +39,7 @@ def register_strategy(name, args=()):
     return wrap
 
 
-def optimize_task_graph(target_task_graph, params, do_not_optimize,
+def optimize_task_graph(target_task_graph, requested_tasks, params, do_not_optimize,
                         decision_task_id, existing_tasks=None, strategy_override=None):
     """
     Perform task optimization, returning a taskgraph and a map from label to
@@ -58,6 +58,7 @@ def optimize_task_graph(target_task_graph, params, do_not_optimize,
 
     removed_tasks = remove_tasks(
         target_task_graph=target_task_graph,
+        requested_tasks=requested_tasks,
         optimizations=optimizations,
         params=params,
         do_not_optimize=do_not_optimize)
@@ -102,7 +103,7 @@ def _log_optimization(verb, opt_counts):
         logger.info('No tasks {} during optimization'.format(verb))
 
 
-def remove_tasks(target_task_graph, params, optimizations, do_not_optimize):
+def remove_tasks(target_task_graph, requested_tasks, params, optimizations, do_not_optimize):
     """
     Implement the "Removing Tasks" phase, returning a set of task labels of all removed tasks.
     """
@@ -124,6 +125,15 @@ def remove_tasks(target_task_graph, params, optimizations, do_not_optimize):
             logger.debug(message.format(label=label, verb=verb, reason="dependent tasks"))
             continue
 
+        # Some tasks in the task graph only exist because they were required
+        # by a task that has just been optimized away. They can now be removed.
+        if label not in requested_tasks:
+            reason = "downstreams-optimized"
+            verb = "removed"
+            removed.add(label)
+            opt_counts['downstreams-optimized'] += 1
+            logger.debug(message.format(label=label, verb=verb, reason=reason))
+
         # call the optimization strategy
         task = target_task_graph.tasks[label]
         opt_by, opt, arg = optimizations(label)