diff --git a/apps/core/lib/core/schema/docker_image.ex b/apps/core/lib/core/schema/docker_image.ex
index 7d1e088e9..00805bece 100644
--- a/apps/core/lib/core/schema/docker_image.ex
+++ b/apps/core/lib/core/schema/docker_image.ex
@@ -24,6 +24,8 @@ defmodule Core.Schema.DockerImage do
     )
   end
 
+  def search(query \\ __MODULE__, s), do: from(di in query, where: like(di.tag, ^"#{s}%"))
+
   def for_tag(query \\ __MODULE__, tag) do
     from(di in query, where: di.tag == ^tag)
   end
diff --git a/apps/core/lib/core/services/scan.ex b/apps/core/lib/core/services/scan.ex
new file mode 100644
index 000000000..61acb4734
--- /dev/null
+++ b/apps/core/lib/core/services/scan.ex
@@ -0,0 +1,80 @@
+defmodule Core.Services.Scan do
+  require Logger
+  alias Core.Services.{Repositories, Versions}
+  alias Core.Schema.{DockerImage, Version, Chart, Terraform}
+  alias Core.Docker.TrivySource
+
+  def scan_image(%DockerImage{} = image) do
+    %{docker_repository: %{repository: repo} = dkr} = img = Core.Repo.preload(image, [docker_repository: :repository])
+    %{publisher: %{owner: owner}} = Core.Repo.preload(repo, [publisher: :owner])
+
+    registry_name = img_name(img)
+    {:ok, registry_token} = Repositories.docker_token([:pull], "#{repo.name}/#{dkr.name}", owner)
+    env = [{"TRIVY_REGISTRY_TOKEN", registry_token} | Core.conf(:docker_env)]
+
+    image = "#{registry_name}:#{image.tag}"
+    Logger.info "Scanning image #{image}"
+    case System.cmd("trivy", ["--quiet", "image", "--format", "json", image, "--timeout", "10m0s"], env: env) do
+      {output, 0} ->
+        case Jason.decode(output) do
+          {:ok, [%{"Vulnerabilities" => vulns} | _]} -> insert_vulns(vulns, img)
+          {:ok, %{"Results" => [_ | _] = res, "SchemaVersion" => 2}} ->
+            Enum.flat_map(res, &Map.get(&1, "Vulnerabilities", []))
+            |> insert_vulns(img)
+            |> log()
+          res ->
+            Logger.info "irregular trivy output #{inspect(res)}"
+            insert_vulns([], img)
+            |> log()
+        end
+      {output, _} ->
+        Logger.info "Trivy failed with: #{output}"
+        handle_trivy_error(output, img)
+    end
+  end
+
+  def terrascan(%Version{} = version) do
+    {type, url} = terrascan_details(version)
+    {output, _} = System.cmd("terrascan", [
+      "scan",
+      "--iac-type", type,
+      "--remote-type", "http",
+      "--remote-url", url,
+      "--output", "json",
+      "--use-colors", "f"
+    ])
+    Versions.record_scan(output, version)
+  end
+
+  defp handle_trivy_error(output, %DockerImage{} = img) do
+    case String.contains?(output, "timeout") do
+      true -> Ecto.Changeset.change(img, %{scan_completed_at: Timex.now()}) |> Core.Repo.update()
+      _ -> :error
+    end
+  end
+
+  defp terrascan_details(%Version{
+    version: v,
+    chart: %Chart{name: chart, repository: %{name: name}}
+  }) do
+    {"helm", "http://chartmuseum:8080/cm/#{name}/charts/#{chart}-#{v}.tgz"}
+  end
+  defp terrascan_details(%Version{terraform: %Terraform{}} = v),
+    do: {"terraform", Core.Storage.url({v.package, v}, :original)}
+
+  defp img_name(%DockerImage{docker_repository: %{repository: repo} = dkr}),
+    do: "#{Core.conf(:registry)}/#{repo.name}/#{dkr.name}"
+
+  defp insert_vulns(nil, img), do: insert_vulns([], img)
+  defp insert_vulns(vulns, img) when is_list(vulns) do
+    Logger.info "found #{length(vulns)} vulnerabilities for #{img_name(img)}"
+    vulns
+    |> Enum.map(&TrivySource.to_vulnerability/1)
+    |> Repositories.add_vulnerabilities(img)
+  end
+
+  defp log({:ok, %DockerImage{id: id, vulnerabilities: vulns}}) when is_list(vulns) do
+    Logger.info "Found #{length(vulns)} vulns for image #{id}"
+  end
+  defp log(_), do: :ok
+end
diff --git a/apps/core/test/services/scan_test.exs b/apps/core/test/services/scan_test.exs
new file mode 100644
index 000000000..fb7b76a7d
--- /dev/null
+++ b/apps/core/test/services/scan_test.exs
@@ -0,0 +1,77 @@
+defmodule Core.Services.ScanTest do
+  use Core.SchemaCase, async: true
+  use Mimic
+
+  alias Core.Services.Scan
+
+  describe "#scan_image/1" do
+    test "it can execute a trivy command" do
+      image = insert(:docker_image)
+      image_name = "dkr.plural.sh/#{image.docker_repository.repository.name}/#{image.docker_repository.name}:#{image.tag}"
+      vuln = Application.get_env(:core, :vulnerability)
+      expect(System, :cmd, fn
+        "trivy", ["--quiet", "image", "--format", "json", ^image_name, "--timeout", "10m0s"], [env: [{"TRIVY_REGISTRY_TOKEN", _}]] ->
+          {~s([{"Vulnerabilities": [#{vuln}]}]), 0}
+      end)
+
+      {:ok, scanned} = Scan.scan_image(image)
+
+      assert scanned.id == image.id
+      assert scanned.grade == :c
+      assert scanned.scan_completed_at
+
+      [vuln] = scanned.vulnerabilities
+      assert vuln.image_id == scanned.id
+    end
+  end
+
+  describe "terrascan/2" do
+    test "it can scan a terraform version" do
+      result = terrascan_res()
+      version = insert(:version,
+        terraform: insert(:terraform, package: %{file_name: "file.tgz", updated_at: nil}),
+        chart: nil,
+        chart_id: nil
+      )
+      url = Core.Storage.url({version.package, version}, :original)
+      expect(System, :cmd, fn
+        "terrascan", [
+          "scan",
+          "--iac-type", "terraform",
+          "--remote-type", "http",
+          "--remote-url", ^url,
+          "--output", "json",
+          "--use-colors", "f"
+        ] ->
+          {result, 0}
+      end)
+
+      {:ok, scanned} = Scan.terrascan(version)
+
+      assert scanned.grade == :d
+    end
+
+    test "it can scan a chart version" do
+      result = terrascan_res()
+      version = insert(:version, chart: insert(:chart))
+      url = "http://chartmuseum:8080/cm/#{version.chart.repository.name}/charts/#{version.chart.name}-#{version.version}.tgz"
+      expect(System, :cmd, fn
+        "terrascan", [
+          "scan",
+          "--iac-type", "helm",
+          "--remote-type", "http",
+          "--remote-url", ^url,
+          "--output", "json",
+          "--use-colors", "f"
+        ] ->
+          {result, 0}
+      end)
+
+      {:ok, scanned} = Scan.terrascan(version)
+
+      assert scanned.grade == :d
+    end
+  end
+
+  defp terrascan_res(), do: priv_file(:core, "scan.json") |> File.read!()
+end
diff --git a/apps/core/test/test_helper.exs b/apps/core/test/test_helper.exs
index 25e98284b..e6349761a 100644
--- a/apps/core/test/test_helper.exs
+++ b/apps/core/test/test_helper.exs
@@ -28,5 +28,6 @@ Mimic.copy(OAuth2.Client)
 Mimic.copy(Core.OAuth.Github)
 Mimic.copy(Core.Services.Shell.Pods)
 Mimic.copy(Vault)
+Mimic.copy(System)
 
 {:ok, _} = Application.ensure_all_started(:ex_machina)
diff --git a/apps/graphql/lib/graphql/resolvers/docker.ex b/apps/graphql/lib/graphql/resolvers/docker.ex
index fb716aff1..c159d93c4 100644
--- a/apps/graphql/lib/graphql/resolvers/docker.ex
+++ b/apps/graphql/lib/graphql/resolvers/docker.ex
@@ -17,6 +17,7 @@ defmodule GraphQl.Resolvers.Docker do
   def list_images(%{docker_repository_id: repo} = args, _) do
     DockerImage.for_repository(repo)
     |> DockerImage.ordered()
+    |> maybe_search(DockerImage, args)
     |> paginate(args)
   end
 
diff --git a/apps/graphql/lib/graphql/schema/docker.ex b/apps/graphql/lib/graphql/schema/docker.ex
index 6901d6a66..c9887ab72 100644
--- a/apps/graphql/lib/graphql/schema/docker.ex
+++ b/apps/graphql/lib/graphql/schema/docker.ex
@@ -104,6 +104,7 @@ defmodule GraphQl.Schema.Docker do
     connection field :docker_images, node_type: :docker_image do
       middleware Authenticated
       arg :docker_repository_id, non_null(:id)
+      arg :q, :string
 
       resolve &Docker.list_images/2
     end
diff --git a/apps/graphql/test/queries/docker_queries_test.exs b/apps/graphql/test/queries/docker_queries_test.exs
index c63a979c0..433f9154c 100644
--- a/apps/graphql/test/queries/docker_queries_test.exs
+++ b/apps/graphql/test/queries/docker_queries_test.exs
@@ -44,6 +44,27 @@ defmodule GraphQl.DockerQueriesTest do
       assert from_connection(found)
              |> ids_equal(docker_imgs)
     end
+
+    test "it can search by tag" do
+      repo = insert(:docker_repository)
+      insert_list(3, :docker_image, docker_repository: repo)
+      img = insert(:docker_image, docker_repository: repo, tag: "3.0")
+
+      {:ok, %{data: %{"dockerImages" => found}}} = run_query("""
+        query DockerImages($id: ID!) {
+          dockerImages(dockerRepositoryId: $id, q: "3." first: 5) {
+            edges {
+              node {
+                id
+              }
+            }
+          }
+        }
+      """, %{"id" => repo.id}, %{current_user: insert(:user)})
+
+      assert from_connection(found)
+             |> ids_equal([img])
+    end
   end
 
   describe "dockerImage" do
diff --git a/apps/worker/lib/worker/conduit/subscribers/docker.ex b/apps/worker/lib/worker/conduit/subscribers/docker.ex
deleted file mode 100644
index 7de5753e0..000000000
--- a/apps/worker/lib/worker/conduit/subscribers/docker.ex
+++ /dev/null
@@ -1,57 +0,0 @@
-defmodule Worker.Conduit.Subscribers.Docker do
-  use Conduit.Subscriber
-  import Conduit.Message
-  alias Core.Docker.TrivySource
-  alias Core.Services.Repositories
-  alias Core.Schema.DockerImage
-  require Logger
-
-  def process(message, _opts) do
-    case scan_image(message.body) do
-      {:ok, result} ->
-        log(result)
-        ack(message)
-      error ->
-        Logger.error "Failed scan: #{inspect(error)}"
-        nack(message)
-    end
-  end
-
-  def scan_image(image) do
-    %{docker_repository: %{repository: repo} = dkr} = img = Core.Repo.preload(image, [docker_repository: :repository])
-    %{publisher: %{owner: owner}} = Core.Repo.preload(repo, [publisher: :owner])
-
-    registry_name  = "#{Worker.conf(:registry)}/#{repo.name}/#{dkr.name}"
-    {:ok, registry_token} = Core.Services.Repositories.docker_token([:pull], "#{repo.name}/#{dkr.name}", owner)
-    env = [{"TRIVY_REGISTRY_TOKEN", registry_token} | Worker.conf(:docker_env)]
-
-    image = "#{registry_name}:#{image.tag}"
-    Logger.info "Scanning image #{image}"
-    case System.cmd("trivy", ["--quiet", "image", "--format", "json", image], env: env) do
-      {output, 0} ->
-        case Jason.decode(output) do
-          {:ok, [%{"Vulnerabilities" => vulns} | _]} -> insert_vulns(vulns, img)
-          {:ok, %{"Results" => [_ | _] = res, "SchemaVersion" => 2}} ->
-            Enum.flat_map(res, &Map.get(&1, "Vulnerabilities", []))
-            |> insert_vulns(img)
-          res ->
-            Logger.info "irregular trivy output #{inspect(res)}"
-            insert_vulns([], img)
-        end
-      {output, _} ->
-        Logger.info "Trivy failed with: #{output}"
-        :error
-    end
-  end
-
-  defp insert_vulns(vulns, img) do
-    (vulns || [])
-    |> Enum.map(&TrivySource.to_vulnerability/1)
-    |> Repositories.add_vulnerabilities(img)
-  end
-
-  defp log(%DockerImage{id: id, vulnerabilities: vulns}) when is_list(vulns) do
-    Logger.info "Found #{length(vulns)} vulns for image #{id}"
-  end
-  defp log(_), do: :ok
-end
diff --git a/apps/worker/lib/worker/conduit/subscribers/scan.ex b/apps/worker/lib/worker/conduit/subscribers/scan.ex
index 67bbb1a15..48a5667ea 100644
--- a/apps/worker/lib/worker/conduit/subscribers/scan.ex
+++ b/apps/worker/lib/worker/conduit/subscribers/scan.ex
@@ -1,9 +1,6 @@
 defmodule Worker.Conduit.Subscribers.Scan do
   use Conduit.Subscriber
   import Conduit.Message
-  alias Core.Services.Versions
-  alias Core.Schema.{Version, Chart, Terraform}
-  require Logger
 
   def process(%Conduit.Message{body: body} = msg, _) do
     case scan(body) do
@@ -12,26 +9,5 @@ defmodule Worker.Conduit.Subscribers.Scan do
     end
   end
 
-  def scan(version) do
-    {type, url} = scan_details(version)
-    {output, _} = System.cmd("terrascan", [
-      "scan",
-      "--iac-type", type,
-      "--remote-type", "http",
-      "--remote-url", url,
-      "--output", "json",
-      "--use-colors", "f"
-    ])
-    Logger.info "terrascan output: #{output}"
-    Versions.record_scan(output, version)
-  end
-
-  defp scan_details(%Version{
-    version: v,
-    chart: %Chart{name: chart, repository: %{name: name}}
-  }) do
-    {"helm", "http://chartmuseum:8080/cm/#{name}/charts/#{chart}-#{v}.tgz"}
-  end
-  defp scan_details(%Version{terraform: %Terraform{}} = v),
-    do: {"terraform", Core.Storage.url({v.package, v}, :original)}
+  def scan(version), do: Core.Services.Scan.terrascan(version)
 end
diff --git a/apps/worker/lib/worker/demo_projects/producer.ex b/apps/worker/lib/worker/demo_projects/producer.ex
index 60e984b95..4c0645d14 100644
--- a/apps/worker/lib/worker/demo_projects/producer.ex
+++ b/apps/worker/lib/worker/demo_projects/producer.ex
@@ -1,13 +1,9 @@
 defmodule Worker.DemoProjects.Producer do
-  use GenStage
-  use Worker.Base
-  require Logger
+  use Worker.PollProducer
   alias Core.Services.Shell.Demo
 
   @max 20
-  @poll :timer.seconds(5)
-
-  defmodule State, do: defstruct [:demand, :draining]
+  @poll Worker.conf(:demo_interval) |>  :timer.seconds()
 
   def start_link(opts \\ []) do
     GenStage.start_link(__MODULE__, opts, name: __MODULE__)
@@ -19,23 +15,11 @@ defmodule Worker.DemoProjects.Producer do
     {:producer, %State{demand: 0}}
   end
 
-  def handle_info(:poll, %State{draining: true} = state), do: empty(state)
-  def handle_info(:poll, %State{demand: demand} = state) do
-    Logger.info "checking for stale demo projects"
-    deliver(demand, %{state | draining: FT.K8S.TrafficDrainHandler.draining?()})
-  end
-
-  def handle_demand(_, %State{draining: true} = state), do: empty(state)
-  def handle_demand(demand, %State{demand: remaining} = state) when demand > 0 do
-    deliver(demand + remaining, state)
-  end
-
-  def handle_demand(_, state), do: empty(state)
-
   defp deliver(demand, state) do
+    Logger.info "checking for expired demo projects"
     case Demo.poll(min(demand, @max)) do
       {:ok, demos} when is_list(demos) ->
-        {:noreply, demos, %{state | demand: demand - length(demos)}}
+        {:noreply, demos, demand(state, demand, demos)}
       _ -> empty(%{state | demand: demand})
     end
   end
diff --git a/apps/worker/lib/worker/docker/pipeline.ex b/apps/worker/lib/worker/docker/pipeline.ex
index 505ed9859..76518946b 100644
--- a/apps/worker/lib/worker/docker/pipeline.ex
+++ b/apps/worker/lib/worker/docker/pipeline.ex
@@ -8,7 +8,7 @@ defmodule Worker.Docker.Pipeline do
       Logger.info "Scheduling docker scan for #{img.id}"
       img
     end)
-    |> Flow.map(&Worker.Conduit.Subscribers.Docker.scan_image/1)
+    |> Flow.map(&Core.Services.Scan.scan_image/1)
     |> Flow.start_link()
   end
 
diff --git a/apps/worker/lib/worker/docker/producer.ex b/apps/worker/lib/worker/docker/producer.ex
index 3e2d70981..648493fea 100644
--- a/apps/worker/lib/worker/docker/producer.ex
+++ b/apps/worker/lib/worker/docker/producer.ex
@@ -1,14 +1,10 @@
 defmodule Worker.Docker.Producer do
-  use GenStage
-  use Worker.Base
-  require Logger
+  use Worker.PollProducer
   alias Core.Services.Repositories
 
   @max 20
   @scan_interval 7
 
-  defmodule State, do: defstruct [:demand, :draining]
-
   def start_link(opts \\ []) do
     GenStage.start_link(__MODULE__, opts, name: __MODULE__)
   end
@@ -19,29 +15,19 @@ defmodule Worker.Docker.Producer do
     {:producer, %State{demand: 0}}
   end
 
-  def handle_info(:poll, %State{draining: true} = state), do: empty(state)
-  def handle_info(:poll, %State{demand: demand} = state),
-    do: deliver(demand, %{state | draining: FT.K8S.TrafficDrainHandler.draining?()})
-
-  def handle_demand(_, %State{draining: true} = state), do: empty(state)
-  def handle_demand(demand, %State{demand: remaining} = state) when demand > 0,
-    do: deliver(demand + remaining, state)
-
-  def handle_demand(_, state), do: empty(state)
-
   defp deliver(demand, state) do
     Logger.info "checking for images to scan, available demand: #{demand}"
     case Repositories.poll_docker_images(@scan_interval, min(demand, @max)) do
       {:ok, imgs} ->
         len = length(imgs)
         Logger.info "found #{len} images"
-        {:noreply, imgs, %{state | demand: demand - len}}
+        {:noreply, imgs, demand(state, demand, len)}
       _ -> empty(%{state | demand: demand})
     end
   end
 
   defp poll_interval() do
-    Core.env("DOCKER_SCAN_POLL_INTERVAL", :int, 60)
+    Core.env("DOCKER_SCAN_POLL_INTERVAL", :int, Worker.conf(:docker_interval))
     |> :timer.seconds()
   end
 end
diff --git a/apps/worker/lib/worker/poll_producer.ex b/apps/worker/lib/worker/poll_producer.ex
new file mode 100644
index 000000000..8a35de4e1
--- /dev/null
+++ b/apps/worker/lib/worker/poll_producer.ex
@@ -0,0 +1,29 @@
+defmodule Worker.PollProducer do
+  defmacro __using__(opts \\ []) do
+    state_keys = Keyword.get(opts, :state, [:demand, :draining])
+
+    quote do
+      use GenStage
+      use Worker.Base
+      require Logger
+
+      defmodule State, do: defstruct unquote(state_keys)
+
+      def handle_info(:poll, %State{draining: true} = state), do: empty(state) |> drain()
+      def handle_info(:poll, %State{demand: 0} = state), do: empty(state) |> drain()
+      def handle_info(:poll, %State{demand: demand} = state),
+        do: deliver(demand, drain(state))
+
+      def handle_demand(_, %State{draining: true} = state), do: empty(state) |> drain()
+      def handle_demand(demand, %State{demand: remaining} = state) when demand > 0,
+        do: deliver(demand + remaining, drain(state))
+
+      def handle_demand(_, state), do: empty(state) |> drain()
+
+      defp drain(state), do: %{state | draining: FT.K8S.TrafficDrainHandler.draining?()}
+
+      defp demand(state, d, v) when is_integer(v), do: %{state | demand: d - v}
+      defp demand(state, d, l) when is_list(l), do: %{state | demand: d - length(l)}
+    end
+  end
+end
diff --git a/apps/worker/lib/worker/rollouts/producer.ex b/apps/worker/lib/worker/rollouts/producer.ex
index 04824e565..c4cbf4fa6 100644
--- a/apps/worker/lib/worker/rollouts/producer.ex
+++ b/apps/worker/lib/worker/rollouts/producer.ex
@@ -1,13 +1,9 @@
 defmodule Worker.Rollouts.Producer do
-  use GenStage
-  use Worker.Base
-  require Logger
+  use Worker.PollProducer
   alias Core.Services.Rollouts
 
   @max 20
-  @poll :timer.seconds(5)
-
-  defmodule State, do: defstruct [:demand, :draining]
+  @poll Worker.conf(:rollout_interval) |> :timer.seconds()
 
   def start_link(opts \\ []) do
     GenStage.start_link(__MODULE__, opts, name: __MODULE__)
@@ -15,27 +11,14 @@ defmodule Worker.Rollouts.Producer do
 
   def init(_) do
     :timer.send_interval(@poll, :poll)
-
     {:producer, %State{demand: 0}}
   end
 
-  def handle_info(:poll, %State{draining: true} = s), do: empty(s)
-  def handle_info(:poll, %State{demand: demand} = state) do
-    Logger.info "checking for new rollouts"
-    deliver(demand, %{state | draining: FT.K8S.TrafficDrainHandler.draining?()})
-  end
-
-  def handle_demand(_, %State{draining: true} = state), do: empty(state)
-  def handle_demand(demand, %State{demand: remaining} = state) when demand > 0 do
-    deliver(demand + remaining, state)
-  end
-
-  def handle_demand(_, state), do: empty(state)
-
   defp deliver(demand, state) do
+    Logger.info "checking for new rollouts"
     case Rollouts.poll(min(demand, @max)) do
       {:ok, rollouts} when is_list(rollouts) ->
-        {:noreply, rollouts, %{state | demand: demand - length(rollouts)}}
+        {:noreply, rollouts, demand(state, demand, rollouts)}
       _ -> empty(%{state | demand: demand})
     end
   end
diff --git a/apps/worker/lib/worker/upgrades/pipeline.ex b/apps/worker/lib/worker/upgrades/pipeline.ex
index 8d25b3544..f67c67dbd 100644
--- a/apps/worker/lib/worker/upgrades/pipeline.ex
+++ b/apps/worker/lib/worker/upgrades/pipeline.ex
@@ -4,7 +4,7 @@ defmodule Worker.Upgrades.Pipeline do
   alias Core.Services.Upgrades
 
   def start_link(producers) do
-    Flow.from_stages(producers, stages: 3, max_demand: 10)
+    Flow.from_stages(producers, stages: 1, max_demand: 10)
     |> Flow.map(fn upgrade ->
       Logger.info "Processing deferred upgrade #{upgrade.id}"
       upgrade
diff --git a/apps/worker/lib/worker/upgrades/producer.ex b/apps/worker/lib/worker/upgrades/producer.ex
index 7719700c2..88794058a 100644
--- a/apps/worker/lib/worker/upgrades/producer.ex
+++ b/apps/worker/lib/worker/upgrades/producer.ex
@@ -1,13 +1,9 @@
 defmodule Worker.Upgrades.Producer do
-  use GenStage
-  use Worker.Base
-  require Logger
+  use Worker.PollProducer, state: [:demand, :type, :draining]
   alias Core.Services.Upgrades
 
   @max 20
-  @poll :timer.seconds(5)
-
-  defmodule State, do: defstruct [:demand, :type, :draining]
+  @poll Worker.conf(:upgrade_interval) |> :timer.seconds()
 
   def start_link(opts \\ []) do
     GenStage.start_link(__MODULE__, opts[:type], name: opts[:name])
@@ -18,19 +14,6 @@ defmodule Worker.Upgrades.Producer do
     {:producer, %State{demand: 0, type: type}}
   end
 
-  def handle_info(:poll, %State{draining: true} = state), do: empty(state)
-  def handle_info(:poll, %State{demand: demand} = state) do
-    Logger.info "checking for new upgrades"
-    deliver(demand, %{state | draining: FT.K8S.TrafficDrainHandler.draining?()})
-  end
-
-  def handle_demand(_, %State{draining: true} = state), do: empty(state)
-  def handle_demand(demand, %State{demand: remaining} = state) when demand > 0 do
-    deliver(demand + remaining, state)
-  end
-
-  def handle_demand(_, state), do: empty(state)
-
   def child_spec(arg) do
     default = %{
       id: arg[:name],
@@ -41,9 +24,10 @@ defmodule Worker.Upgrades.Producer do
   end
 
   defp deliver(demand, %State{type: type} = state) do
+    Logger.info "checking for deferred updates"
     case Upgrades.poll_deferred_updates(type, min(demand, @max)) do
       {:ok, updates} when is_list(updates) ->
-        {:noreply, updates, %{state | demand: demand - length(updates)}}
+        {:noreply, updates, demand(state, demand, updates)}
       _ -> empty(%{state | demand: demand})
     end
   end
diff --git a/apps/worker/test/conduit/subscribers/docker_test.exs b/apps/worker/test/conduit/subscribers/docker_test.exs
deleted file mode 100644
index 542eb04a6..000000000
--- a/apps/worker/test/conduit/subscribers/docker_test.exs
+++ /dev/null
@@ -1,26 +0,0 @@
-defmodule Worker.Conduit.Subscribers.DockerTest do
-  use Core.SchemaCase, async: true
-  use Mimic
-  alias Worker.Conduit.Subscribers.Docker
-
-  describe "#scan_image/1" do
-    test "it can execute a trivy command" do
-      image = insert(:docker_image)
-      image_name = "dkr.plural.sh/#{image.docker_repository.repository.name}/#{image.docker_repository.name}:#{image.tag}"
-      vuln = Application.get_env(:core, :vulnerability)
-      expect(System, :cmd, fn
-        "trivy", ["--quiet", "image", "--format", "json", ^image_name], [env: [{"TRIVY_REGISTRY_TOKEN", _}]] ->
-          {~s([{"Vulnerabilities": [#{vuln}]}]), 0}
-      end)
-
-      {:ok, scanned} = Docker.scan_image(image)
-
-      assert scanned.id == image.id
-      assert scanned.grade == :c
-      assert scanned.scan_completed_at
-
-      [vuln] = scanned.vulnerabilities
-      assert vuln.image_id == scanned.id
-    end
-  end
-end
diff --git a/apps/worker/test/demo_projects/pipeline_test.exs b/apps/worker/test/demo_projects/pipeline_test.exs
index e90cf9c88..2a12830e3 100644
--- a/apps/worker/test/demo_projects/pipeline_test.exs
+++ b/apps/worker/test/demo_projects/pipeline_test.exs
@@ -17,7 +17,7 @@ defmodule Worker.DemoProjects.PipelineTest do
 
       demo = insert(:demo_project, inserted_at: Timex.now() |> Timex.shift(hours: -24))
 
-      :timer.sleep(:timer.seconds(6))
+      :timer.sleep(:timer.seconds(3))
 
       refute refetch(demo)
     end
diff --git a/apps/worker/test/docker/pipeline_test.exs b/apps/worker/test/docker/pipeline_test.exs
index 05351024c..e26773b62 100644
--- a/apps/worker/test/docker/pipeline_test.exs
+++ b/apps/worker/test/docker/pipeline_test.exs
@@ -12,12 +12,12 @@ defmodule Worker.Docker.PipelineTest do
       insert(:docker_image, scanned_at: Timex.now(), scan_completed_at: Timex.now())
 
       me = self()
-      expect(Worker.Conduit.Subscribers.Docker, :scan_image, 3, fn img -> send me, {:dkr, img} end)
+      expect(Core.Services.Scan, :scan_image, 3, fn img -> send me, {:dkr, img} end)
 
       {:ok, producer} = Docker.Producer.start_link()
       {:ok, _} = Docker.Pipeline.start_link(producer)
 
-      :timer.sleep(:timer.seconds(6))
+      :timer.sleep(:timer.seconds(3))
 
       found = Enum.map(1..3, fn _ ->
         assert_receive {:dkr, img}
diff --git a/apps/worker/test/rollouts/pipeline_test.exs b/apps/worker/test/rollouts/pipeline_test.exs
index 9802acdb8..f98b32526 100644
--- a/apps/worker/test/rollouts/pipeline_test.exs
+++ b/apps/worker/test/rollouts/pipeline_test.exs
@@ -11,7 +11,7 @@ defmodule Worker.Rollouts.PipelineTest do
       event = %PubSub.VersionCreated{item: insert(:version)}
       roll = insert(:rollout, status: :queued, repository: build(:repository), event: event)
 
-      :timer.sleep(:timer.seconds(6))
+      :timer.sleep(:timer.seconds(3))
 
       assert refetch(roll).status == :finished
     end
diff --git a/apps/worker/test/test_helper.exs b/apps/worker/test/test_helper.exs
index 3325c3013..e5acb416f 100644
--- a/apps/worker/test/test_helper.exs
+++ b/apps/worker/test/test_helper.exs
@@ -3,6 +3,6 @@ Mimic.copy(Goth.Token)
 Mimic.copy(GoogleApi.CloudResourceManager.V3.Api.Projects)
 Mimic.copy(Cloudflare.DnsRecord)
 Mimic.copy(Worker.Conduit.Broker)
-Mimic.copy(Worker.Conduit.Subscribers.Docker)
+Mimic.copy(Core.Services.Scan)
 
 ExUnit.start()
diff --git a/apps/worker/test/upgrades/pipeline_test.exs b/apps/worker/test/upgrades/pipeline_test.exs
index 3a03fcd90..d1dcbe0f7 100644
--- a/apps/worker/test/upgrades/pipeline_test.exs
+++ b/apps/worker/test/upgrades/pipeline_test.exs
@@ -24,7 +24,7 @@ defmodule Worker.Upgrades.PipelineTest do
       ]})
       deferred = insert(:deferred_update, chart_installation: chart_inst, version: version, user: user)
 
-      :timer.sleep(:timer.seconds(6))
+      :timer.sleep(:timer.seconds(3))
 
       refute refetch(deferred)
       assert refetch(chart_inst).version_id == version.id
diff --git a/config/config.exs b/config/config.exs
index e5fbe0edc..fdac4db08 100644
--- a/config/config.exs
+++ b/config/config.exs
@@ -123,7 +123,8 @@ config :core,
   onplural_domain: "onplural.sh",
   gcp_organization: "1323",
   gcp_identity: "someone@example.com",
-  vault: "https://vault.vault:8201"
+  vault: "https://vault.vault:8201",
+  docker_env: []
 
 config :briefly,
   directory: [{:system, "TMPDIR"}, {:system, "TMP"}, {:system, "TEMP"}, "/tmp"],
@@ -149,4 +150,10 @@ config :hammer,
   backend: {Hammer.Backend.ETS, [expiry_ms: 60_000 * 60,
                                  cleanup_interval_ms: 60_000 * 10]}
 
+config :worker,
+  upgrade_interval: 10,
+  demo_interval: 10,
+  rollout_interval: 10,
+  docker_interval: 60
+
 import_config "#{Mix.env()}.exs"
diff --git a/config/prod.exs b/config/prod.exs
index 2f9de68af..361a49e87 100644
--- a/config/prod.exs
+++ b/config/prod.exs
@@ -44,6 +44,12 @@ config :worker, docker_env: [
   # {"DOCKER_TLS_VERIFY", "1"},
 ]
 
+config :core, docker_env: [
+  # {"DOCKER_HOST", "tcp://localhost:2376"},
+  # {"DOCKER_CERT_PATH", "/certs/client"},
+  # {"DOCKER_TLS_VERIFY", "1"},
+]
+
 config :ex_aws,
   region: {:system, "AWS_REGION"},
   secret_access_key: [{:system, "AWS_ACCESS_KEY_ID"}, {:awscli, "profile_name", 30}],
diff --git a/config/test.exs b/config/test.exs
index 8db65d110..e34d3306d 100644
--- a/config/test.exs
+++ b/config/test.exs
@@ -81,6 +81,11 @@ config :worker,
 config :worker,
   upgrade_pipeline: []
 
+config :worker,
+  upgrade_interval: 1,
+  demo_interval: 1,
+  rollout_interval: 1,
+  docker_interval: 1
 
 config :core,
   workos_webhook: "supersecret"
diff --git a/plural/helm/plural/Chart.yaml b/plural/helm/plural/Chart.yaml
index d61816202..23ace0059 100644
--- a/plural/helm/plural/Chart.yaml
+++ b/plural/helm/plural/Chart.yaml
@@ -2,7 +2,7 @@ apiVersion: v2
 name: plural
 description: A helm chart for installing plural
 appVersion: "0.8.10"
-version: 0.8.92
+version: 0.8.93
 dependencies:
 - name: hydra
   repository: https://k8s.ory.sh/helm/charts
diff --git a/plural/helm/plural/templates/deployment.yaml b/plural/helm/plural/templates/deployment.yaml
index c201b78a9..ce223ef87 100644
--- a/plural/helm/plural/templates/deployment.yaml
+++ b/plural/helm/plural/templates/deployment.yaml
@@ -107,6 +107,9 @@ spec:
             name: plural-env
         env:
         {{ include "plural.env" . | nindent 8 }}
+        {{- if .Values.worker.extraEnv }}
+        {{ .Values.worker.extraEnv | toYaml | nindent 8 }}
+        {{- end }}
         - name: DOCKER_SCAN_POLL_INTERVAL
           value: {{ .Values.worker.dockerScan.pollInterval | quote }}
         - name: DOCKER_SCAN_PARALLELISM
diff --git a/plural/helm/plural/values.yaml b/plural/helm/plural/values.yaml
index 6566cb2c9..000e337ec 100644
--- a/plural/helm/plural/values.yaml
+++ b/plural/helm/plural/values.yaml
@@ -393,6 +393,8 @@ worker:
 
   nodeSelector: {}
 
+  extraEnv: []
+
   tolerations:
   - key: plural.sh/pluralReserved
     operator: Exists