diff --git a/changelog/650.feature.rst b/changelog/650.feature.rst
new file mode 100644
index 0000000..ce6dc39
--- /dev/null
+++ b/changelog/650.feature.rst
@@ -0,0 +1 @@
+Added new ``pytest_handlecrashitem`` hook to allow handling and rescheduling crashed items.
diff --git a/src/xdist/dsession.py b/src/xdist/dsession.py
index ab927fa..ccd5e20 100644
--- a/src/xdist/dsession.py
+++ b/src/xdist/dsession.py
@@ -343,6 +343,12 @@ class DSession:
             nodeid, (fspath, None, fspath), (), "failed", msg, "???"
         )
         rep.node = worker
+
+        self.config.hook.pytest_handlecrashitem(
+            crashitem=nodeid,
+            report=rep,
+            sched=self.sched,
+        )
         self.config.hook.pytest_runtest_logreport(report=rep)
 
 
diff --git a/src/xdist/newhooks.py b/src/xdist/newhooks.py
index da0f22a..a6443f3 100644
--- a/src/xdist/newhooks.py
+++ b/src/xdist/newhooks.py
@@ -64,3 +64,20 @@ def pytest_xdist_auto_num_workers(config):
 
     .. versionadded:: 2.1
     """
+
+
+@pytest.mark.firstresult
+def pytest_handlecrashitem(crashitem, report, sched):
+    """
+    Handle a crashitem, modifying the report if necessary.
+
+    The scheduler is provided as a parameter to reschedule the test if desired with
+    `sched.mark_test_pending`.
+
+    def pytest_handlecrashitem(crashitem, report, sched):
+        if should_rerun(crashitem):
+            sched.mark_test_pending(crashitem)
+            report.outcome = "rerun"
+
+    .. versionadded:: 2.2.1
+    """
diff --git a/src/xdist/scheduler/each.py b/src/xdist/scheduler/each.py
index b2a0442..cfe99e7 100644
--- a/src/xdist/scheduler/each.py
+++ b/src/xdist/scheduler/each.py
@@ -101,6 +101,14 @@ class EachScheduling:
     def mark_test_complete(self, node, item_index, duration=0):
         self.node2pending[node].remove(item_index)
 
+    def mark_test_pending(self, item):
+        self.pending.insert(
+            0,
+            self.collection.index(item),
+        )
+        for node in self.node2pending:
+            self.check_schedule(node)
+
     def remove_node(self, node):
         # KeyError if we didn't get an add_node() yet
         pending = self.node2pending.pop(node)
diff --git a/src/xdist/scheduler/load.py b/src/xdist/scheduler/load.py
index e378d9a..f32caa5 100644
--- a/src/xdist/scheduler/load.py
+++ b/src/xdist/scheduler/load.py
@@ -151,6 +151,14 @@ class LoadScheduling:
         self.node2pending[node].remove(item_index)
         self.check_schedule(node, duration=duration)
 
+    def mark_test_pending(self, item):
+        self.pending.insert(
+            0,
+            self.collection.index(item),
+        )
+        for node in self.node2pending:
+            self.check_schedule(node)
+
     def check_schedule(self, node, duration=0):
         """Maybe schedule new items on the node
 
diff --git a/src/xdist/scheduler/loadscope.py b/src/xdist/scheduler/loadscope.py
index 31dbe26..c25e476 100644
--- a/src/xdist/scheduler/loadscope.py
+++ b/src/xdist/scheduler/loadscope.py
@@ -243,6 +243,9 @@ class LoadScopeScheduling:
         self.assigned_work[node][scope][nodeid] = True
         self._reschedule(node)
 
+    def mark_test_pending(self, item):
+        raise NotImplementedError()
+
     def _assign_work_unit(self, node):
         """Assign a work unit to a node."""
         assert self.workqueue
diff --git a/testing/test_newhooks.py b/testing/test_newhooks.py
index 0318442..d2c2878 100644
--- a/testing/test_newhooks.py
+++ b/testing/test_newhooks.py
@@ -60,3 +60,37 @@ class TestHooks:
             ["*HOOK: gw0 test_a, test_b, test_c", "*HOOK: gw1 test_a, test_b, test_c"]
         )
         res.stdout.fnmatch_lines(["*3 passed*"])
+
+
+class TestCrashItem:
+    @pytest.fixture(autouse=True)
+    def create_test_file(self, testdir):
+        testdir.makepyfile(
+            """
+            import os
+            def test_a(): pass
+            def test_b(): os._exit(1)
+            def test_c(): pass
+            def test_d(): pass
+        """
+        )
+
+    def test_handlecrashitem(self, testdir):
+        """Test pytest_handlecrashitem hook."""
+        testdir.makeconftest(
+            """
+            test_runs = 0
+
+            def pytest_handlecrashitem(crashitem, report, sched):
+                global test_runs
+
+                if test_runs == 0:
+                    sched.mark_test_pending(crashitem)
+                    test_runs = 1
+                else:
+                    print("HOOK: pytest_handlecrashitem")
+        """
+        )
+        res = testdir.runpytest("-n2", "-s")
+        res.stdout.fnmatch_lines_random(["*HOOK: pytest_handlecrashitem"])
+        res.stdout.fnmatch_lines(["*3 passed*"])