diff --git a/lib/mix/lib/mix/lock.ex b/lib/mix/lib/mix/lock.ex
new file mode 100644
index 00000000000..855bb891633
--- /dev/null
+++ b/lib/mix/lib/mix/lock.ex
@@ -0,0 +1,322 @@
+defmodule Mix.Lock do
+  @moduledoc false
+
+  # Lock implementation working across multiple OS processes.
+  #
+  # The lock is implemented using TCP sockets and hard links.
+  #
+  # A process holds the lock if it owns a TCP socket, whose port is
+  # written in the lock_0 file. We need to create such lock files
+  # atomically, so the process first writes its port to a port_P
+  # file and then attempts to create a hard link to it at lock_0.
+  #
+  # An inherent problem with lock files is that the lock owner may
+  # terminate abruptly, leaving a "stale" file. Other processes can
+  # detect a stale file by reading the port written in that file,
+  # trying to connect to that port and failing. In order for another
+  # process to link to the same path, the file needs to be replaced.
+  # However, we need to guarantee that only a single process can
+  # remove or replace the file, otherwise a concurrent process may
+  # end up removing a newly linked file.
+  #
+  # To address this problem we employ a chained locking procedure.
+  # Specifically, we attempt to link our port to lock_0, if that
+  # fails, we try to connect to the lock_0 port. If we manage to
+  # connect, it means the lock is taken, so we wait for it to close
+  # and start over. If we fail to connect, it means the lock is stale,
+  # so we want to replace it. In order to do that, we try to obtain
+  # lock_1. Again, we try to link and connect. Eventually, we should
+  # successfully link to lock_N. At that point we can clean up all
+  # the files, so we perform these steps:
+  #
+  #   * move our port_P to lock_0
+  #   * remove all the other port_P files
+  #   * remove all lock_1+ files
+  #
+  # It is important to perform these steps in this order, to avoid
+  # race conditions. By moving to lock_0, we make sure that all new
+  # processes trying to lock will connect to our port. By removing
+  # all port_P files we make sure that currently paused processes
+  # that are about to link port_P at lock_N will fail to link, since
+  # the port_P file will no longer exist (once lock_N is removed).
+  #
+  # Finally, note that we do not remove the lock file in `unlock/1`.
+  # If we did that, another process could try to connect and fail
+  # because the file would not exist, in such case the process would
+  # assume the file is stale and needs to be replaced, therefore
+  # possibly replacing another process who successfully links at the
+  # empty spot. This means we effectively always leave a stale file,
+  # however, in order to shortcut the port check for future processes,
+  # we atomically replace the file content with port 0, to indicate
+  # the file is stale.
+  #
+  # The main caveat of using ephemeral TCP ports is that they are not
+  # unique. This creates a theoretical scenario where the lock holder
+  # terminates abruptly and leaves its port in lock_0, then the port
+  # is assigned to a unrelated process (unaware of the locking). To
+  # handle this scenario, when we connect to a lock_N port, we expect
+  # it to immediately send us `@probe_data`. If this does not happen
+  # within `@probe_timeout_ms`, we assume the port is taken by an
+  # unrelated process and the lock file is stale. Note that it is ok
+  # to use a long timeout, because this scenario is very unlikely.
+  # Theoretically, if an actual lock owner is not able to send the
+  # probe data within the timeout, the lock will fail, however with
+  # a high enough timeout, this should not be a problem in practice.
+
+  @loopback {127, 0, 0, 1}
+  @listen_opts [:binary, ip: @loopback, packet: :raw, nodelay: true, backlog: 128, active: false]
+  @connect_opts [:binary, packet: :raw, nodelay: true, active: false]
+  @probe_data "elixirlock"
+  @probe_data_size byte_size(@probe_data)
+  @probe_timeout_ms 5_000
+
+  @doc """
+  Acquires a lock identified by the given key.
+
+  This function blocks until the lock is acquired by this process,
+  and then executes `fun`, returning its return value.
+
+  This function can also be called if this process already has the
+  lock. In such case the function is executed immediately.
+
+  ## Options
+
+    * `:on_taken` - a one-arity function called if the lock is held
+      by a different process. The operating system PID of that process
+      is given as the first argument (as a string). This function may
+      be called multiple times, if the lock owner changes, until it
+      is successfully acquired by this process.
+
+  """
+  @spec with_lock(iodata(), (-> term()), keyword()) :: term()
+  def with_lock(key, fun, opts \\ []) do
+    opts = Keyword.validate!(opts, [:on_taken])
+
+    key = key |> :erlang.md5() |> Base.url_encode64(padding: false)
+    path = Path.join([System.tmp_dir!(), "mix_lock", key])
+
+    pdict_key = {__MODULE__, path}
+    has_lock? = Process.get(pdict_key)
+
+    if has_lock? do
+      fun.()
+    else
+      lock = lock(path, opts[:on_taken])
+      Process.put(pdict_key, true)
+
+      try do
+        fun.()
+      after
+        # Unlocking will always close the socket, but it may raise,
+        # so we remove key from the dictionary first
+        Process.delete(pdict_key)
+        unlock(lock)
+      end
+    end
+  end
+
+  defp lock(path, on_taken) do
+    File.mkdir_p!(path)
+
+    case listen() do
+      {:ok, socket, port} ->
+        spawn_link(fn -> accept_loop(socket) end)
+
+        try do
+          try_lock(path, socket, port, on_taken)
+        rescue
+          exception ->
+            # Close the socket to make sure we don't block the lock
+            :gen_tcp.close(socket)
+            reraise exception, __STACKTRACE__
+        end
+
+      {:error, reason} ->
+        Mix.raise(
+          "failed to open a TCP socket while acquiring a lock, reason: #{inspect(reason)}"
+        )
+    end
+  end
+
+  defp listen do
+    with {:ok, socket} <- :gen_tcp.listen(0, @listen_opts) do
+      case :inet.port(socket) do
+        {:ok, port} ->
+          {:ok, socket, port}
+
+        {:error, reason} ->
+          :gen_tcp.close(socket)
+          {:error, reason}
+      end
+    end
+  end
+
+  defp try_lock(path, socket, port, on_taken) do
+    port_path = Path.join(path, "port_#{port}")
+    os_pid = System.pid()
+
+    File.write!(port_path, <<port::unsigned-integer-32, os_pid::binary>>, [:raw])
+
+    case grab_lock(path, port_path, 0) do
+      {:ok, 0} ->
+        # We grabbed lock_0, so all good
+        %{socket: socket, path: path}
+
+      {:ok, _n} ->
+        # We grabbed lock_1+, so we need to replace lock_0 and clean up
+        take_over(path, port_path)
+        %{socket: socket, path: path}
+
+      {:taken, probe_socket, os_pid} ->
+        # Another process has the lock, wait for close and start over
+        if on_taken, do: on_taken.(os_pid)
+        await_close(probe_socket)
+        try_lock(path, socket, port, on_taken)
+
+      :invalidated ->
+        try_lock(path, socket, port, on_taken)
+    end
+  end
+
+  defp grab_lock(path, port_path, n) do
+    lock_path = Path.join(path, "lock_#{n}")
+
+    case File.ln(port_path, lock_path) do
+      :ok ->
+        {:ok, n}
+
+      {:error, :eexist} ->
+        case probe(lock_path) do
+          {:ok, probe_socket, os_pid} ->
+            {:taken, probe_socket, os_pid}
+
+          {:error, _reason} ->
+            grab_lock(path, port_path, n + 1)
+        end
+
+      {:error, :enoent} ->
+        :invalidated
+
+      {:error, reason} ->
+        raise File.LinkError,
+          reason: reason,
+          action: "create hard link",
+          existing: port_path,
+          new: lock_path
+    end
+  end
+
+  defp accept_loop(listen_socket) do
+    case :gen_tcp.accept(listen_socket) do
+      {:ok, socket} ->
+        _ = :gen_tcp.send(socket, @probe_data)
+        accept_loop(listen_socket)
+
+      # eintr is "Interrupted system call".
+      {:error, :eintr} ->
+        accept_loop(listen_socket)
+
+      {:error, reason} when reason in [:closed, :einval] ->
+        :ok
+    end
+  end
+
+  defp probe(port_path) do
+    with {:ok, port, os_pid} <- fetch_probe_port(port_path),
+         {:ok, socket} <- connect(port),
+         {:ok, socket} <- await_probe_data(socket) do
+      {:ok, socket, os_pid}
+    end
+  end
+
+  defp fetch_probe_port(port_path) do
+    case File.read(port_path) do
+      {:ok, <<0::unsigned-integer-32>>} -> {:error, :ignore}
+      {:ok, <<port::unsigned-integer-32, os_pid::binary>>} -> {:ok, port, os_pid}
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
+  defp connect(port) do
+    # On Windows connecting to an unbound port takes a few seconds to
+    # fail, so instead we shortcut the check by attempting a listen,
+    # which succeeds or fails immediately. Note that `reuseaddr` here
+    # ensures that if the listening socket closed recently, we can
+    # immediately reclaim the same port.
+    case :gen_tcp.listen(port, [reuseaddr: true] ++ @listen_opts) do
+      {:ok, socket} ->
+        :gen_tcp.close(socket)
+        # The port is free, so connecting would fail
+        {:error, :econnrefused}
+
+      {:error, _reason} ->
+        :gen_tcp.connect(@loopback, port, @connect_opts)
+    end
+  end
+
+  defp await_probe_data(socket) do
+    case :gen_tcp.recv(socket, @probe_data_size, @probe_timeout_ms) do
+      {:ok, @probe_data} ->
+        {:ok, socket}
+
+      {:ok, _data} ->
+        :gen_tcp.close(socket)
+        {:error, :unexpected_port_owner}
+
+      {:error, :eintr} ->
+        await_probe_data(socket)
+
+      {:error, reason} ->
+        :gen_tcp.close(socket)
+        {:error, reason}
+    end
+  end
+
+  defp take_over(path, port_path) do
+    # The operations here must happen in precise order, so if anything
+    # fails, we keep the files as is and the next process that grabs
+    # the lock will do the cleanup
+
+    lock_path = Path.join(path, "lock_0")
+
+    # We linked to lock_N successfully, so port_path should exist
+    File.rename!(port_path, lock_path)
+
+    names = File.ls!(path)
+
+    for "port_" <> _ = name <- names do
+      File.rm!(Path.join(path, name))
+    end
+
+    for "lock_" <> _ = name <- names, name != "lock_0" do
+      File.rm!(Path.join(path, name))
+    end
+  end
+
+  defp await_close(socket) do
+    case :gen_tcp.recv(socket, 0) do
+      {:error, :closed} ->
+        :ok
+
+      {:error, :eintr} ->
+        await_close(socket)
+
+      {:error, _other} ->
+        # In case of an unexpected error, we close the socket ourselves
+        # to retry
+        :gen_tcp.close(socket)
+    end
+  end
+
+  defp unlock(lock) do
+    port_path = Path.join(lock.path, "port_0")
+    lock_path = Path.join(lock.path, "lock_0")
+
+    File.write!(port_path, <<0::unsigned-integer-32>>, [:raw])
+    File.rename!(port_path, lock_path)
+  after
+    # Closing the socket will cause the accepting process to finish
+    # and all accepted sockets (tied to that process) will get closed
+    :gen_tcp.close(lock.socket)
+  end
+end
diff --git a/lib/mix/lib/mix/project.ex b/lib/mix/lib/mix/project.ex
index 6d9988d8ee9..d7ae332754d 100644
--- a/lib/mix/lib/mix/project.ex
+++ b/lib/mix/lib/mix/project.ex
@@ -882,6 +882,23 @@ defmodule Mix.Project do
     end
   end
 
+  @doc false
+  def with_build_lock(config \\ config(), fun) do
+    # To avoid duplicated compilation, we wrap compilation tasks, such
+    # as compile.all, deps.compile, compile.elixir, compile.erlang in
+    # a lock. Note that compile.all covers compile.elixir, but the
+    # latter can still be invoked directly, so we put the lock over
+    # each individual task.
+
+    build_path = build_path(config)
+
+    on_taken = fn os_pid ->
+      Mix.shell().info("Waiting for lock on the build directory (held by process #{os_pid})")
+    end
+
+    Mix.Lock.with_lock(build_path, fun, on_taken: on_taken)
+  end
+
   # Loads mix.exs in the current directory or loads the project from the
   # mixfile cache and pushes the project onto the project stack.
   defp load_project(app, post_config) do
diff --git a/lib/mix/lib/mix/state.ex b/lib/mix/lib/mix/state.ex
index 41e6744b08a..5c99f13f9b6 100644
--- a/lib/mix/lib/mix/state.ex
+++ b/lib/mix/lib/mix/state.ex
@@ -9,15 +9,6 @@ defmodule Mix.State do
     GenServer.start_link(__MODULE__, :ok, name: @name)
   end
 
-  def lock(key, fun) do
-    try do
-      GenServer.call(@name, {:lock, key}, @timeout)
-      fun.()
-    after
-      GenServer.call(@name, {:unlock, key}, @timeout)
-    end
-  end
-
   def builtin_apps do
     GenServer.call(@name, :builtin_apps, @timeout)
   end
@@ -83,8 +74,6 @@ defmodule Mix.State do
     )
 
     state = %{
-      key_to_waiting: %{},
-      pid_to_key: %{},
       builtin_apps: :code.get_path()
     }
 
@@ -114,69 +103,6 @@ defmodule Mix.State do
     end
   end
 
-  @impl true
-  def handle_call({:lock, key}, {pid, _} = from, state) do
-    %{key_to_waiting: key_to_waiting, pid_to_key: pid_to_key} = state
-
-    key_to_waiting =
-      case key_to_waiting do
-        %{^key => {locked, waiting}} ->
-          Map.put(key_to_waiting, key, {locked, :queue.in(from, waiting)})
-
-        %{} ->
-          go!(from)
-          Map.put(key_to_waiting, key, {pid, :queue.new()})
-      end
-
-    ref = Process.monitor(pid)
-    pid_to_key = Map.put(pid_to_key, pid, {key, ref})
-    {:noreply, %{state | key_to_waiting: key_to_waiting, pid_to_key: pid_to_key}}
-  end
-
-  @impl true
-  def handle_call({:unlock, key}, {pid, _}, state) do
-    %{key_to_waiting: key_to_waiting, pid_to_key: pid_to_key} = state
-    {{^key, ref}, pid_to_key} = Map.pop(pid_to_key, pid)
-    Process.demonitor(ref, [:flush])
-    key_to_waiting = unlock(key_to_waiting, pid_to_key, key)
-    {:reply, :ok, %{state | key_to_waiting: key_to_waiting, pid_to_key: pid_to_key}}
-  end
-
-  @impl true
-  def handle_info({:DOWN, ref, _type, pid, _reason}, state) do
-    %{key_to_waiting: key_to_waiting, pid_to_key: pid_to_key} = state
-    {{key, ^ref}, pid_to_key} = Map.pop(pid_to_key, pid)
-
-    key_to_waiting =
-      case key_to_waiting do
-        %{^key => {^pid, _}} ->
-          unlock(key_to_waiting, pid_to_key, key)
-
-        %{^key => {locked, waiting}} ->
-          waiting = :queue.delete_with(fn {qpid, _qref} -> qpid == pid end, waiting)
-          Map.put(key_to_waiting, key, {locked, waiting})
-      end
-
-    {:noreply, %{state | key_to_waiting: key_to_waiting, pid_to_key: pid_to_key}}
-  end
-
-  defp unlock(key_to_waiting, pid_to_key, key) do
-    %{^key => {_locked, waiting}} = key_to_waiting
-
-    case :queue.out(waiting) do
-      {{:value, {pid, _} = from}, waiting} ->
-        # Assert that we still know this PID
-        _ = Map.fetch!(pid_to_key, pid)
-        go!(from)
-        Map.put(key_to_waiting, key, {pid, waiting})
-
-      {:empty, _waiting} ->
-        Map.delete(key_to_waiting, key)
-    end
-  end
-
-  defp go!(from), do: GenServer.reply(from, :ok)
-
   # ../elixir/ebin -> elixir
   # ../ssl-9.6/ebin -> ssl
   defp app_from_code_path(path) do
diff --git a/lib/mix/lib/mix/tasks/compile.all.ex b/lib/mix/lib/mix/tasks/compile.all.ex
index 04930d769ab..c4b73477bda 100644
--- a/lib/mix/lib/mix/tasks/compile.all.ex
+++ b/lib/mix/lib/mix/tasks/compile.all.ex
@@ -12,8 +12,15 @@ defmodule Mix.Tasks.Compile.All do
   @impl true
   def run(args) do
     Mix.Project.get!()
+
     config = Mix.Project.config()
 
+    Mix.Project.with_build_lock(config, fn ->
+      do_run(config, args)
+    end)
+  end
+
+  defp do_run(config, args) do
     # Compute the app cache if it is stale and we are
     # not compiling from a dependency.
     app_cache =
diff --git a/lib/mix/lib/mix/tasks/compile.elixir.ex b/lib/mix/lib/mix/tasks/compile.elixir.ex
index 2b2d67d7739..7ae75848361 100644
--- a/lib/mix/lib/mix/tasks/compile.elixir.ex
+++ b/lib/mix/lib/mix/tasks/compile.elixir.ex
@@ -123,11 +123,10 @@ defmodule Mix.Tasks.Compile.Elixir do
       |> profile_opts()
 
     # Having compilations racing with other is most undesired,
-    # so we wrap the compiler in a lock. Ideally we would use
-    # flock in the future.
+    # so we wrap the compiler in a lock.
 
     with_logger_app(project, fn ->
-      Mix.State.lock(__MODULE__, fn ->
+      Mix.Project.with_build_lock(project, fn ->
         Mix.Compilers.Elixir.compile(
           manifest,
           srcs,
diff --git a/lib/mix/lib/mix/tasks/compile.erlang.ex b/lib/mix/lib/mix/tasks/compile.erlang.ex
index 301331b99ac..4667e15a71b 100644
--- a/lib/mix/lib/mix/tasks/compile.erlang.ex
+++ b/lib/mix/lib/mix/tasks/compile.erlang.ex
@@ -46,11 +46,15 @@ defmodule Mix.Tasks.Compile.Erlang do
   @impl true
   def run(args) do
     {opts, _, _} = OptionParser.parse(args, switches: @switches)
+
     project = Mix.Project.config()
-    source_paths = project[:erlc_paths]
-    Mix.Compilers.Erlang.assert_valid_erlc_paths(source_paths)
-    files = Mix.Utils.extract_files(source_paths, [:erl])
-    do_run(files, opts, project, source_paths)
+
+    Mix.Project.with_build_lock(project, fn ->
+      source_paths = project[:erlc_paths]
+      Mix.Compilers.Erlang.assert_valid_erlc_paths(source_paths)
+      files = Mix.Utils.extract_files(source_paths, [:erl])
+      do_run(files, opts, project, source_paths)
+    end)
   end
 
   defp do_run([], _, _, _), do: {:noop, []}
diff --git a/lib/mix/lib/mix/tasks/deps.compile.ex b/lib/mix/lib/mix/tasks/deps.compile.ex
index 0865990c6e9..89bed61743e 100644
--- a/lib/mix/lib/mix/tasks/deps.compile.ex
+++ b/lib/mix/lib/mix/tasks/deps.compile.ex
@@ -53,15 +53,20 @@ defmodule Mix.Tasks.Deps.Compile do
     end
 
     Mix.Project.get!()
-    deps = Mix.Dep.load_and_cache()
 
-    case OptionParser.parse(args, switches: @switches) do
-      {opts, [], _} ->
-        compile(filter_available_and_local_deps(deps), opts)
+    config = Mix.Project.config()
 
-      {opts, tail, _} ->
-        compile(Mix.Dep.filter_by_name(tail, deps, opts), opts)
-    end
+    Mix.Project.with_build_lock(config, fn ->
+      deps = Mix.Dep.load_and_cache()
+
+      case OptionParser.parse(args, switches: @switches) do
+        {opts, [], _} ->
+          compile(filter_available_and_local_deps(deps), opts)
+
+        {opts, tail, _} ->
+          compile(Mix.Dep.filter_by_name(tail, deps, opts), opts)
+      end
+    end)
   end
 
   @doc false
diff --git a/lib/mix/test/mix/lock_test.exs b/lib/mix/test/mix/lock_test.exs
new file mode 100644
index 00000000000..1c53b0217ff
--- /dev/null
+++ b/lib/mix/test/mix/lock_test.exs
@@ -0,0 +1,187 @@
+Code.require_file("../test_helper.exs", __DIR__)
+
+defmodule Mix.LockTest do
+  use ExUnit.Case, async: true
+
+  @lock_key Atom.to_string(__MODULE__)
+
+  test "executes functions" do
+    assert Mix.Lock.with_lock(@lock_key, fn -> :it_works! end) == :it_works!
+    assert Mix.Lock.with_lock(@lock_key, fn -> :still_works! end) == :still_works!
+  end
+
+  test "releases lock on error" do
+    assert_raise RuntimeError, fn ->
+      Mix.Lock.with_lock(@lock_key, fn -> raise "oops" end)
+    end
+
+    assert Mix.Lock.with_lock(@lock_key, fn -> :still_works! end) == :still_works!
+  end
+
+  test "releases lock on exit" do
+    {_pid, ref} =
+      spawn_monitor(fn ->
+        Mix.Lock.with_lock(@lock_key, fn -> Process.exit(self(), :kill) end)
+      end)
+
+    assert_receive {:DOWN, ^ref, _, _, _}
+    assert Mix.Lock.with_lock(@lock_key, fn -> :still_works! end) == :still_works!
+  end
+
+  test "blocks until released" do
+    parent = self()
+
+    task =
+      Task.async(fn ->
+        Mix.Lock.with_lock(@lock_key, fn ->
+          send(parent, :locked)
+          assert_receive :will_lock
+          :it_works!
+        end)
+      end)
+
+    assert_receive :locked
+    send(task.pid, :will_lock)
+    assert Mix.Lock.with_lock(@lock_key, fn -> :still_works! end) == :still_works!
+    assert Task.await(task) == :it_works!
+  end
+
+  @tag :capture_log
+  test "blocks until released on error" do
+    parent = self()
+
+    {pid, ref} =
+      spawn_monitor(fn ->
+        Mix.Lock.with_lock(@lock_key, fn ->
+          send(parent, :locked)
+          assert_receive :will_lock
+          raise "oops"
+        end)
+      end)
+
+    assert_receive :locked
+    send(pid, :will_lock)
+    assert Mix.Lock.with_lock(@lock_key, fn -> :still_works! end) == :still_works!
+    assert_receive {:DOWN, ^ref, _, _, _}
+  end
+
+  test "blocks until released on exit" do
+    parent = self()
+
+    {pid, ref} =
+      spawn_monitor(fn ->
+        Mix.Lock.with_lock(@lock_key, fn ->
+          send(parent, :locked)
+          assert_receive :will_not_lock
+        end)
+      end)
+
+    assert_receive :locked
+    Process.exit(pid, :kill)
+    assert Mix.Lock.with_lock(@lock_key, fn -> :still_works! end) == :still_works!
+    assert_receive {:DOWN, ^ref, _, _, _}
+  end
+
+  test "scheduls and releases on exit" do
+    assert Mix.Lock.with_lock(@lock_key, fn ->
+             {pid, ref} =
+               spawn_monitor(fn ->
+                 Mix.Lock.with_lock(@lock_key, fn ->
+                   raise "this will never be invoked"
+                 end)
+               end)
+
+             Process.exit(pid, :kill)
+             assert_receive {:DOWN, ^ref, _, _, :killed}
+             :it_works!
+           end) == :it_works!
+
+    assert Mix.Lock.with_lock(@lock_key, fn -> :still_works! end) == :still_works!
+  end
+
+  @tag :tmp_dir
+  test "property test with file access", %{tmp_dir: tmp_dir} do
+    # Spawn N concurrent processes incrementing number in a file
+    n = 10
+    number_path = Path.join(tmp_dir, "number.txt")
+
+    File.write!(number_path, "0")
+
+    refs =
+      for _ <- 1..n do
+        spawn_monitor(fn ->
+          Mix.Lock.with_lock(@lock_key, fn ->
+            number = number_path |> File.read!() |> String.to_integer()
+            new_number = number + 1
+            File.write!(number_path, Integer.to_string(new_number))
+
+            assert File.read!(number_path) == Integer.to_string(new_number)
+
+            # Terminate without unlocking in random cases
+            case Enum.random(1..2) do
+              1 -> Process.exit(self(), :kill)
+              2 -> :ok
+            end
+          end)
+        end)
+        |> elem(1)
+      end
+
+    await_monitors(refs)
+
+    assert File.read!(number_path) == Integer.to_string(n)
+  end
+
+  test "lock can be acquired multiple times by the same process" do
+    {_pid, ref} =
+      spawn_monitor(fn ->
+        Mix.Lock.with_lock(@lock_key, fn ->
+          Mix.Lock.with_lock(@lock_key, fn ->
+            Process.exit(self(), :kill)
+          end)
+        end)
+      end)
+
+    assert_receive {:DOWN, ^ref, _, _, _}
+
+    assert Mix.Lock.with_lock(@lock_key, fn ->
+             Mix.Lock.with_lock(@lock_key, fn ->
+               :still_works!
+             end)
+           end) == :still_works!
+  end
+
+  test "calls :on_taken when the lock is held by a different process" do
+    parent = self()
+
+    {pid, ref} =
+      spawn_monitor(fn ->
+        Mix.Lock.with_lock(@lock_key, fn ->
+          send(parent, :locked)
+          assert_receive :will_lock
+        end)
+      end)
+
+    assert_receive :locked
+
+    on_taken = fn os_pid ->
+      send(pid, :will_lock)
+      send(self(), {:on_taken_called, os_pid})
+    end
+
+    assert Mix.Lock.with_lock(@lock_key, fn -> :it_works! end, on_taken: on_taken) == :it_works!
+
+    os_pid = System.pid()
+    assert_receive {:on_taken_called, ^os_pid}
+
+    assert_receive {:DOWN, ^ref, _, _, _}
+  end
+
+  defp await_monitors([]), do: :ok
+
+  defp await_monitors(refs) do
+    receive do
+      {:DOWN, ref, _, _, _} -> await_monitors(refs -- [ref])
+    end
+  end
+end
diff --git a/lib/mix/test/mix/state_test.exs b/lib/mix/test/mix/state_test.exs
deleted file mode 100644
index aaf131bec82..00000000000
--- a/lib/mix/test/mix/state_test.exs
+++ /dev/null
@@ -1,101 +0,0 @@
-Code.require_file("../test_helper.exs", __DIR__)
-
-defmodule Mix.StateTest do
-  use ExUnit.Case, async: true
-
-  describe "lock" do
-    test "executes functions" do
-      assert Mix.State.lock(:key, fn -> :it_works! end) == :it_works!
-      assert Mix.State.lock(:key, fn -> :still_works! end) == :still_works!
-    end
-
-    test "releases lock on error" do
-      assert_raise RuntimeError, fn ->
-        Mix.State.lock(:key, fn -> raise "oops" end)
-      end
-
-      assert Mix.State.lock(:key, fn -> :still_works! end) == :still_works!
-    end
-
-    test "releases lock on exit" do
-      {_pid, ref} =
-        spawn_monitor(fn ->
-          Mix.State.lock(:key, fn -> Process.exit(self(), :kill) end)
-        end)
-
-      assert_receive {:DOWN, ^ref, _, _, _}
-      assert Mix.State.lock(:key, fn -> :still_works! end) == :still_works!
-    end
-
-    test "blocks until released" do
-      parent = self()
-
-      task =
-        Task.async(fn ->
-          Mix.State.lock(:key, fn ->
-            send(parent, :locked)
-            assert_receive :will_lock
-            :it_works!
-          end)
-        end)
-
-      assert_receive :locked
-      send(task.pid, :will_lock)
-      assert Mix.State.lock(:key, fn -> :still_works! end) == :still_works!
-      assert Task.await(task) == :it_works!
-    end
-
-    @tag :capture_log
-    test "blocks until released on error" do
-      parent = self()
-
-      {pid, ref} =
-        spawn_monitor(fn ->
-          Mix.State.lock(:key, fn ->
-            send(parent, :locked)
-            assert_receive :will_lock
-            raise "oops"
-          end)
-        end)
-
-      assert_receive :locked
-      send(pid, :will_lock)
-      assert Mix.State.lock(:key, fn -> :still_works! end) == :still_works!
-      assert_receive {:DOWN, ^ref, _, _, _}
-    end
-
-    test "blocks until released on exit" do
-      parent = self()
-
-      {pid, ref} =
-        spawn_monitor(fn ->
-          Mix.State.lock(:key, fn ->
-            send(parent, :locked)
-            assert_receive :will_not_lock
-          end)
-        end)
-
-      assert_receive :locked
-      Process.exit(pid, :kill)
-      assert Mix.State.lock(:key, fn -> :still_works! end) == :still_works!
-      assert_receive {:DOWN, ^ref, _, _, _}
-    end
-
-    test "scheduls and releases on exit" do
-      assert Mix.State.lock(:key, fn ->
-               {pid, ref} =
-                 spawn_monitor(fn ->
-                   Mix.State.lock(:key, fn ->
-                     raise "this will never be invoked"
-                   end)
-                 end)
-
-               Process.exit(pid, :kill)
-               assert_receive {:DOWN, ^ref, _, _, :killed}
-               :it_works!
-             end) == :it_works!
-
-      assert Mix.State.lock(:key, fn -> :still_works! end) == :still_works!
-    end
-  end
-end