diff --git a/.github/workflows/scripts/pytest-selected.py b/.github/workflows/scripts/pytest-selected.py
index 65bd6116fb40..6fa86e75d215 100644
--- a/.github/workflows/scripts/pytest-selected.py
+++ b/.github/workflows/scripts/pytest-selected.py
@@ -22,6 +22,7 @@
 import json
 import math
 import os
+import shlex
 import sys
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -161,8 +162,31 @@ def file_weights(files: list[str]) -> dict[str, float]:
     return weights
 
 
+def per_file_plan_applies(args: list[str], files: list[str]) -> bool:
+    """Whether planning by file gives each shard the tests pytest would run.
+
+    `-m`/`-k` filter after the plan, so most shards could end up with no
+    selected tests (e.g. `-m snuba_ci` in Snuba's CI). The sentry plugin
+    matches the planned paths relative to the rootdir, which
+    pytest_shard_plugin checks is the working directory.
+    """
+    every_arg = [*args, *shlex.split(os.environ.get("PYTEST_ADDOPTS", ""))]
+    if any(arg.startswith(("-m", "-k")) and not arg.startswith("--") for arg in every_arg):
+        return False
+    return not any(os.path.isabs(f) or f.startswith(os.pardir + os.sep) for f in files)
+
+
 def run_full_shard(args: list[str], total: int, group: int) -> int:
     files = list_test_files(args)
+    if not per_file_plan_applies(args, files):
+        # As before per-file sharding: every shard collects everything and the
+        # sentry plugin keeps its hash share of the items.
+        print(
+            "pytest-selected: -m/-k or paths outside the working directory, so "
+            "every shard collects everything and keeps its hash share",
+            file=sys.stderr,
+        )
+        return int(pytest.main(args))
     plan = plan_full_shards(file_weights(files), total) if files else None
     if plan is None:
         return int(pytest.main(args))
diff --git a/.github/workflows/scripts/pytest_shard_plugin.py b/.github/workflows/scripts/pytest_shard_plugin.py
index 959ddddceb52..9e9a60e2d79d 100644
--- a/.github/workflows/scripts/pytest_shard_plugin.py
+++ b/.github/workflows/scripts/pytest_shard_plugin.py
@@ -11,6 +11,7 @@
 import json
 import os
 from hashlib import sha256
+from pathlib import Path
 
 import pytest
 
@@ -19,6 +20,16 @@ def item_part(nodeid: str, parts: int) -> int:
     return int(sha256(nodeid.encode()).hexdigest(), 16) % parts
 
 
+def pytest_configure(config: pytest.Config) -> None:
+    # pytest-selected.py writes paths relative to the working directory, and
+    # the sentry plugin matches them relative to the rootdir.
+    if os.environ.get("PYTEST_SHARD_PARTS") and config.rootpath != Path.cwd().resolve():
+        raise pytest.UsageError(
+            f"pytest_shard_plugin: rootdir {config.rootpath} is not the working "
+            f"directory {Path.cwd()}; run pytest-selected.py from the rootdir"
+        )
+
+
 def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
     parts_file = os.environ.get("PYTEST_SHARD_PARTS")
     if not parts_file:
diff --git a/.github/workflows/scripts/report-backend-test-failures.js b/.github/workflows/scripts/report-backend-test-failures.js
index 87352f95ad06..de645913e0b1 100644
--- a/.github/workflows/scripts/report-backend-test-failures.js
+++ b/.github/workflows/scripts/report-backend-test-failures.js
@@ -42,12 +42,28 @@ export function parseFailures(files, core) {
       .filter(t => t.outcome === 'failed')
       .map(t => ({
         nodeid: t.nodeid ?? 'unknown',
-        longrepr: (t.call ?? t.setup ?? t.teardown ?? {}).longrepr ?? '',
+        longrepr:
+          failedSubtests(t) || ((t.call ?? t.setup ?? t.teardown ?? {}).longrepr ?? ''),
         artifactDir,
       }));
   });
 }
 
+// The parent's own longrepr only says "contains N failed subtests". Render
+// each failed subtest the way pytest's SUBFAILED line does.
+function failedSubtests(test) {
+  return (test.subtests ?? [])
+    .filter(s => s.outcome === 'failed')
+    .map(({context, longrepr}) => {
+      const kwargs = Object.entries(context?.kwargs ?? {}).map(([k, v]) => `${k}=${v}`);
+      const msg =
+        context?.msg === null || context?.msg === undefined ? [] : [`[${context.msg}]`];
+      const description = [...msg, ...(kwargs.length ? [`(${kwargs.join(', ')})`] : [])];
+      return `SUBFAILED ${description.join(' ') || '(<subtest>)'}\n${longrepr ?? ''}`;
+    })
+    .join('\n\n');
+}
+
 // Maps artifact directory names to GitHub Actions job log URLs.
 const JOB_MATCHERS = [
   {
diff --git a/.github/workflows/scripts/report-backend-test-failures.test.js b/.github/workflows/scripts/report-backend-test-failures.test.js
index 05be483fb99c..7a93c2a06aa4 100644
--- a/.github/workflows/scripts/report-backend-test-failures.test.js
+++ b/.github/workflows/scripts/report-backend-test-failures.test.js
@@ -238,6 +238,34 @@ describe('parseFailures', () => {
 
     assert.equal(parseFailures([file1, file2], mockCore()).length, 2);
   });
+
+  it('uses failed subtests instead of the parent summary', () => {
+    const file = createArtifactFile(
+      'pytest-results-backend-111-0',
+      'pytest.json',
+      makePytestJson([
+        {
+          nodeid: 'tests/a.py::test_a',
+          outcome: 'failed',
+          call: {outcome: 'failed', longrepr: 'contains 1 failed subtest'},
+          subtests: [
+            {outcome: 'passed', context: {msg: null, kwargs: {n: '1'}}},
+            {
+              outcome: 'failed',
+              longrepr: 'AssertionError: boom',
+              context: {msg: 'm', kwargs: {f: '<function f>', n: '2'}},
+            },
+          ],
+        },
+      ])
+    );
+
+    const [failure] = parseFailures([file], mockCore());
+    assert.equal(
+      failure.longrepr,
+      'SUBFAILED [m] (f=<function f>, n=2)\nAssertionError: boom'
+    );
+  });
 });
 
 describe('extractNodeids', () => {
diff --git a/.github/workflows/scripts/test_backend_test_durations.py b/.github/workflows/scripts/test_backend_test_durations.py
index 95097fadd47b..adaa2f24620e 100644
--- a/.github/workflows/scripts/test_backend_test_durations.py
+++ b/.github/workflows/scripts/test_backend_test_durations.py
@@ -2,12 +2,13 @@
 
 import importlib
 import json
+from pathlib import Path
 
 # Module has a hyphen in its name, so use importlib.
 _mod = importlib.import_module("backend-test-durations")
 
 
-def test_extract_sums_stages_per_file(tmp_path):
+def test_extract_sums_stages_per_file(tmp_path: Path) -> None:
     report = tmp_path / "pytest.json"
     report.write_text(
         json.dumps(
@@ -18,6 +19,9 @@ def test_extract_sums_stages_per_file(tmp_path):
                         "setup": {"duration": 1.0},
                         "call": {"duration": 2.0},
                         "teardown": {"duration": 0.5},
+                        # Subtests ran inside the call; reruns don't count.
+                        "subtests": [{"duration": 1.5}],
+                        "reruns": [{"call": {"duration": 9.0}}],
                     },
                     {"nodeid": "tests/a.py::test_2[x]", "setup": {"duration": 0.5}},
                     {"nodeid": "tests/b.py::test_3", "call": {"duration": 4.0}},
@@ -28,7 +32,7 @@ def test_extract_sums_stages_per_file(tmp_path):
     assert _mod.extract(report) == {"files": {"tests/a.py": 4.0, "tests/b.py": 4.0}, "tests": 3}
 
 
-def test_merge_adds_up_shards():
+def test_merge_adds_up_shards() -> None:
     shards = [
         {"files": {"tests/a.py": 1.04, "tests/b.py": 2.0}, "tests": 2},
         {"files": {"tests/a.py": 1.0}, "tests": 2},
diff --git a/.github/workflows/scripts/test_pytest_selected.py b/.github/workflows/scripts/test_pytest_selected.py
index 7582c0d0a4b5..4c90a31fc0a5 100644
--- a/.github/workflows/scripts/test_pytest_selected.py
+++ b/.github/workflows/scripts/test_pytest_selected.py
@@ -4,6 +4,9 @@
 import importlib
 import json
 import os
+from contextlib import AbstractContextManager
+from pathlib import Path
+from typing import Any
 from unittest import mock
 
 import pytest
@@ -16,14 +19,14 @@
 
 
 class TestPlanShards:
-    def test_packs_whole_files_by_weight(self):
+    def test_packs_whole_files_by_weight(self) -> None:
         # 10 + 1 vs 6 + 5
         assert plan_shards({"a.py": 10, "b.py": 6, "c.py": 5, "d.py": 1}, 2) == [
             Shard(["a.py", "d.py"]),
             Shard(["b.py", "c.py"]),
         ]
 
-    def test_splits_a_file_too_big_for_one_shard(self):
+    def test_splits_a_file_too_big_for_one_shard(self) -> None:
         # 80 tests over 4 shards: big.py is worth 3 shards, the rest share one.
         assert plan_shards({"a.py": 10, "big.py": 60, "b.py": 10}, 4) == [
             Shard(["a.py", "b.py"]),
@@ -32,18 +35,18 @@ def test_splits_a_file_too_big_for_one_shard(self):
             Shard(["big.py"], 3, 2),
         ]
 
-    def test_spare_shards_go_to_the_heaviest_file(self):
+    def test_spare_shards_go_to_the_heaviest_file(self) -> None:
         assert plan_shards({"a.py": 3, "b.py": 1}, 3) == [
             Shard(["b.py"]),
             Shard(["a.py"], 2, 0),
             Shard(["a.py"], 2, 1),
         ]
 
-    def test_keeps_a_shard_for_the_small_files(self):
+    def test_keeps_a_shard_for_the_small_files(self) -> None:
         # big.py rounds up to both shards on its own.
         assert plan_shards({"big.py": 3, "a.py": 1}, 2) == [Shard(["big.py"]), Shard(["a.py"])]
 
-    def test_every_file_runs_exactly_once(self):
+    def test_every_file_runs_exactly_once(self) -> None:
         weights = {f"t{i}.py": i % 7 + 1 for i in range(37)}
         plan = plan_shards(weights, 5)
         assert len(plan) == 5
@@ -51,14 +54,21 @@ def test_every_file_runs_exactly_once(self):
         assert sorted(f for shard in plan for f in shard.files) == sorted(weights)
 
 
-def run_main(tmp_path, selected, counts, total, group, exit_code=pytest.ExitCode.OK):
+def run_main(
+    tmp_path: Path,
+    selected: list[str],
+    counts: dict[str, int],
+    total: int,
+    group: int,
+    exit_code: pytest.ExitCode = pytest.ExitCode.OK,
+) -> tuple[int, list[tuple[object, ...]]]:
     selected_file = tmp_path / "selected-tests.txt"
     selected_file.write_text("".join(f"{path}\n" for path in selected))
     if counts:
         (tmp_path / "selected-tests.counts.json").write_text(json.dumps(counts))
-    calls = []
+    calls: list[tuple[object, ...]] = []
 
-    def fake_main(args):
+    def fake_main(args: list[str]) -> pytest.ExitCode:
         env = ("SELECTED_TESTS_FILE", "TOTAL_TEST_GROUPS", "TEST_GROUP")
         calls.append((args, *(os.environ[k] for k in env)))
         return exit_code
@@ -73,23 +83,23 @@ def fake_main(args):
 
 
 class TestMain:
-    def test_runs_only_this_shards_files(self, tmp_path):
+    def test_runs_only_this_shards_files(self, tmp_path: Path) -> None:
         exit_code, calls = run_main(tmp_path, ["a.py", "b.py"], {"a.py": 5, "b.py": 3}, 2, 1)
         shard_file = tmp_path / "selected-tests.shard-1.txt"
         assert exit_code == 0
         assert calls == [(["tests", "-q"], str(shard_file), "1", "0")]
         assert shard_file.read_text() == "b.py\n"
 
-    def test_split_file_shards_by_test(self, tmp_path):
+    def test_split_file_shards_by_test(self, tmp_path: Path) -> None:
         _, calls = run_main(tmp_path, ["a.py", "big.py"], {"a.py": 1, "big.py": 9}, 3, 2)
         assert calls[0][2:] == ("2", "1")
         assert (tmp_path / "selected-tests.shard-2.txt").read_text() == "big.py\n"
 
-    def test_single_shard_runs_pytest_unchanged(self, tmp_path):
+    def test_single_shard_runs_pytest_unchanged(self, tmp_path: Path) -> None:
         _, calls = run_main(tmp_path, ["a.py"], {}, 1, 0)
         assert calls == [(["tests", "-q"], str(tmp_path / "selected-tests.txt"), "1", "0")]
 
-    def test_empty_shard_is_not_a_failure(self, tmp_path):
+    def test_empty_shard_is_not_a_failure(self, tmp_path: Path) -> None:
         exit_code, _ = run_main(
             tmp_path, ["a.py", "b.py"], {}, 2, 0, exit_code=pytest.ExitCode.NO_TESTS_COLLECTED
         )
@@ -100,19 +110,19 @@ def test_empty_shard_is_not_a_failure(self, tmp_path):
 item_part = importlib.import_module("pytest_shard_plugin").item_part
 
 
-def pieces_per_shard(n):
+def pieces_per_shard(n: int) -> AbstractContextManager[object]:
     return mock.patch.object(_mod, "PIECES_PER_SHARD", n)
 
 
-def write_files(root, files):
+def write_files(root: Path, files: list[str]) -> None:
     for path in files:
         (root / path).parent.mkdir(parents=True, exist_ok=True)
         (root / path).write_text("def test_one(): pass\ndef test_two(): pass\n")
 
 
-def assert_runs_every_test_once(plan, tests):
+def assert_runs_every_test_once(plan: list[Any], tests: dict[str, list[str]]) -> None:
     """`tests` maps each file to its test ids. Checks each one runs on exactly one shard."""
-    ran = []
+    ran: list[str] = []
     for shard in plan:
         for path in shard.files:
             part, parts = shard.parts.get(path, (0, 1))
@@ -121,7 +131,7 @@ def assert_runs_every_test_once(plan, tests):
 
 
 class TestPlanFullShards:
-    def test_packs_whole_files_by_weight(self):
+    def test_packs_whole_files_by_weight(self) -> None:
         weights = {f"t{i}.py": 1.0 for i in range(24)} | {"z.py": 0.0}
         with pieces_per_shard(1):
             plan = plan_full_shards(weights, 4)
@@ -129,7 +139,7 @@ def test_packs_whole_files_by_weight(self):
         assert [len(s.files) for s in plan] == [7, 6, 6, 6]
         assert all(not s.parts for s in plan)
 
-    def test_splits_slow_files_across_shards(self):
+    def test_splits_slow_files_across_shards(self) -> None:
         # Shards hold 24s, so files over 2s are cut into 2s parts.
         weights = {"big.py": 10.0} | {f"t{i}.py": 1.0 for i in range(86)}
         with pieces_per_shard(12):
@@ -139,19 +149,19 @@ def test_splits_slow_files_across_shards(self):
         assert sorted(s.parts["big.py"] for s in holders) == [(i, 4) for i in range(4)]
         assert all(len(s.parts) == 1 for s in plan)
 
-    def test_never_splits_a_file_past_the_shard_count(self):
+    def test_never_splits_a_file_past_the_shard_count(self) -> None:
         plan = plan_full_shards({"big.py": 100.0, "a.py": 1.0, "b.py": 1.0}, 3)
         assert plan is not None
         assert sorted(s.parts["big.py"] for s in plan) == [(0, 3), (1, 3), (2, 3)]
 
-    def test_no_plan_when_a_shard_would_be_empty(self):
+    def test_no_plan_when_a_shard_would_be_empty(self) -> None:
         assert plan_full_shards({"a.py": 0.0}, 2) is None
 
-    def test_same_weights_same_plan(self):
+    def test_same_weights_same_plan(self) -> None:
         weights = {f"t{i}.py": float(i % 13) for i in range(200)}
         assert plan_full_shards(weights, 7) == plan_full_shards(dict(reversed(weights.items())), 7)
 
-    def test_balances_shards(self):
+    def test_balances_shards(self) -> None:
         weights = {f"t{i}.py": float(i % 13 + 1) ** 2 for i in range(500)} | {"big.py": 900.0}
         plan = plan_full_shards(weights, 10)
         assert plan is not None
@@ -160,7 +170,7 @@ def test_balances_shards(self):
         ]
         assert max(loads) - min(loads) <= 1.0
 
-    def test_every_test_runs_exactly_once(self):
+    def test_every_test_runs_exactly_once(self) -> None:
         tests = {f"t{i}.py": [f"t{i}.py::test_{j}" for j in range(i % 50)] for i in range(300)}
         tests["big.py"] = [f"big.py::Test::test_{j}[{k}]" for j in range(40) for k in range(30)]
         weights = {path: len(ids) * 0.5 for path, ids in tests.items()}
@@ -170,7 +180,7 @@ def test_every_test_runs_exactly_once(self):
             assert_runs_every_test_once(plan, tests)
 
 
-def test_list_test_files_lists_python_files_under_roots_minus_ignores(tmp_path):
+def test_list_test_files_lists_python_files_under_roots_minus_ignores(tmp_path: Path) -> None:
     for path in (
         "tests/a/test_a.py",
         "tests/a/helpers.py",
@@ -187,19 +197,31 @@ def test_list_test_files_lists_python_files_under_roots_minus_ignores(tmp_path):
         assert _mod.list_test_files(args) == ["tests/a/helpers.py", "tests/a/test_a.py"]
 
 
-def run_full_main(tmp_path, files, durations, total, group):
+def run_full_main(
+    tmp_path: Path,
+    files: list[str],
+    durations: dict[str, object] | None,
+    total: int,
+    group: int,
+    addopts: str = "",
+) -> tuple[int, list[tuple[object, ...]]]:
     write_files(tmp_path, files)
     durations_file = tmp_path / "durations.json"
     if durations is not None:
         durations_file.write_text(json.dumps(durations))
-    calls = []
+    calls: list[tuple[object, ...]] = []
 
-    def fake_main(args):
+    def fake_main(args: list[str]) -> pytest.ExitCode:
         env = ("SELECTED_TESTS_FILE", "TOTAL_TEST_GROUPS", "TEST_GROUP", "PYTEST_SHARD_PARTS")
         calls.append((args, *(os.environ.get(k) for k in env)))
         return pytest.ExitCode.OK
 
-    env = {"SELECTED_TESTS_FILE": "", "TOTAL_TEST_GROUPS": str(total), "TEST_GROUP": str(group)}
+    env = {
+        "SELECTED_TESTS_FILE": "",
+        "TOTAL_TEST_GROUPS": str(total),
+        "TEST_GROUP": str(group),
+        "PYTEST_ADDOPTS": addopts,
+    }
     with (
         contextlib.chdir(tmp_path),
         mock.patch.dict(os.environ, env),
@@ -210,7 +232,7 @@ def fake_main(args):
 
 
 class TestFullRunMain:
-    def test_runs_only_this_shards_files(self, tmp_path):
+    def test_runs_only_this_shards_files(self, tmp_path: Path) -> None:
         files = ["tests/test_a.py", "tests/test_b.py", "tests/test_new.py"]
         durations = {"seconds_per_test": 1.0, "files": {"tests/test_a.py": 3.0}}
         with pieces_per_shard(1):
@@ -223,7 +245,7 @@ def test_runs_only_this_shards_files(self, tmp_path):
         assert (tmp_path / shard_file).read_text() == "tests/test_b.py\ntests/test_new.py\n"
         assert json.loads((tmp_path / parts_file).read_text()) == {}
 
-    def test_new_and_renamed_files_run_once_with_estimated_weight(self, tmp_path):
+    def test_new_and_renamed_files_run_once_with_estimated_weight(self, tmp_path: Path) -> None:
         # test_renamed.py was test_old.py when the durations were recorded: the stale
         # entry is ignored and the new name is weighted by its 2 tests, like any new file.
         files = [f"tests/test_{i}.py" for i in range(8)] + ["tests/test_renamed.py"]
@@ -246,6 +268,33 @@ def test_new_and_renamed_files_run_once_with_estimated_weight(self, tmp_path):
         assert plan is not None
         assert sorted(f for s in plan for f in s.files) == sorted(files)
 
-    def test_without_durations_every_shard_collects_everything(self, tmp_path):
+    def test_without_durations_every_shard_collects_everything(self, tmp_path: Path) -> None:
         _, calls = run_full_main(tmp_path, ["tests/test_a.py"], None, 3, 1)
         assert calls == [(["tests", "-q"], "", "3", "1", None)]
+
+    def test_filtered_run_falls_back_to_hash_sharding(self, tmp_path: Path) -> None:
+        durations = {"seconds_per_test": 1.0, "files": {}}
+        _, calls = run_full_main(
+            tmp_path, ["tests/test_a.py", "tests/test_b.py"], durations, 16, 12, "-m snuba_ci"
+        )
+        assert calls == [(["tests", "-q"], "", "16", "12", None)]
+
+
+@pytest.mark.parametrize(
+    ("args", "addopts", "files", "applies"),
+    [
+        (["tests"], "--reruns=5 -n 3 --dist=loadfile", ["tests/a.py"], True),
+        (["tests", "--max-worker-restart=1"], "", ["tests/a.py"], True),
+        (["tests", "-m", "snuba_ci"], "", ["tests/a.py"], False),
+        (["tests", "-msnuba_ci"], "", ["tests/a.py"], False),
+        (["tests", "-k", "not slow"], "", ["tests/a.py"], False),
+        (["tests"], "-k 'not __In' -m snuba_ci", ["tests/a.py"], False),
+        (["/abs/tests"], "", ["/abs/tests/a.py"], False),
+        (["../tests"], "", ["../tests/a.py"], False),
+    ],
+)
+def test_per_file_plan_applies(
+    args: list[str], addopts: str, files: list[str], applies: bool
+) -> None:
+    with mock.patch.dict(os.environ, {"PYTEST_ADDOPTS": addopts}):
+        assert _mod.per_file_plan_applies(args, files) is applies
diff --git a/src/sentry/testutils/pytest/json_report_reruns.py b/src/sentry/testutils/pytest/json_report_reruns.py
index 386e611847f6..1972fa4275c2 100644
--- a/src/sentry/testutils/pytest/json_report_reruns.py
+++ b/src/sentry/testutils/pytest/json_report_reruns.py
@@ -1,21 +1,34 @@
-"""Extend the pytest-json-report plugin with a new "reruns" attribute.
+"""Extend the pytest-json-report plugin with "reruns" and "subtests" attributes.
 
 Each test will (if it's run more than once) store its prior results in a
-`reruns` array attribute, where each item may have `setup` `call` and
-`teardown` sections.
+`reruns` array attribute, where each item may have `setup` `call` `teardown`
+and `subtests` sections.
+
+Subtest reports share the parent's nodeid and `when == "call"`, so they are
+kept in a `subtests` array instead of looking like reruns of the call.
 
 See also:
     * https://github.com/pytest-dev/pytest-rerunfailures
     * https://github.com/numirias/pytest-json-report
+    * https://docs.pytest.org/en/stable/how-to/subtests.html
     * https://getsentry.atlassian.net/browse/DEVINFRA-630
 """
 
+from __future__ import annotations
+
+from collections import defaultdict
+from collections.abc import Generator
+from typing import Any
+
 import pytest
 import pytest_jsonreport.plugin
+from _pytest._io.saferepr import saferepr
 
-JSONTestItem = dict[str, list[object]]
+JSONTestItem = dict[str, Any]
 JSONTestItems = dict[str, JSONTestItem]
 
+PHASES = ("setup", "call", "teardown")
+
 
 def pytest_plugin_registered(plugin: object, manager: pytest.PytestPluginManager) -> None:
     """Only register if/when the plugin we're extending is registered."""
@@ -27,12 +40,48 @@ def pytest_plugin_registered(plugin: object, manager: pytest.PytestPluginManager
 class PytestRerunJSONReporter:
     def __init__(self, json_tests: JSONTestItems):
         self.json_tests = json_tests
+        # rerunfailures logs an attempt's reports when it ends, but subtest
+        # reports arrive live, so hold them until their attempt's call report.
+        self.pending_subtests: defaultdict[str, list[dict[str, Any]]] = defaultdict(list)
+
+    @pytest.hookimpl(wrapper=True)
+    def pytest_runtest_logreport(
+        self, report: pytest.TestReport
+    ) -> Generator[None, object, object]:
+        if not isinstance(report, pytest.SubtestReport):
+            return (yield)
+
+        # JSONReport stores the subtest as the call stage and its outcome as
+        # "subtests passed", so put both back.
+        before = self.json_tests.get(report.nodeid, {})
+        prior = {"outcome": "passed"} | {
+            k: before[k] for k in (report.when, "outcome") if k in before
+        }
+        result = yield
+        json_testitem = self.json_tests[report.nodeid]
+        stage = json_testitem.pop(report.when)
+        json_testitem.update(prior)
+        # On an xdist controller the parent's report doesn't count failed subtests.
+        if report.failed:
+            json_testitem["outcome"] = "failed"
+
+        context = report.context
+        self.pending_subtests[report.nodeid].append(
+            {
+                **stage,
+                "context": {
+                    "msg": context.msg,
+                    "kwargs": {str(k): saferepr(v) for k, v in context.kwargs.items()},
+                },
+            }
+        )
+        return result
 
     def pytest_json_runtest_stage(self, report: pytest.TestReport) -> None:
-        assert self.json_tests is not None
+        if isinstance(report, pytest.SubtestReport):
+            return
 
-        nodeid = report.nodeid
-        json_testitem = self.json_tests[nodeid]
+        json_testitem = self.json_tests[report.nodeid]
         if report.when in json_testitem:
             # this is a new result of some kind -- record all prior data as a
             # "rerun" and start over fresh
@@ -40,7 +89,11 @@ def pytest_json_runtest_stage(self, report: pytest.TestReport) -> None:
             reruns.append(
                 {
                     key: json_testitem.pop(key)
-                    for key in ("setup", "call", "teardown")
+                    for key in (*PHASES, "subtests")
                     if key in json_testitem
                 }
             )
+        if report.when == "call":
+            subtests = self.pending_subtests.pop(report.nodeid, None)
+            if subtests:
+                json_testitem["subtests"] = subtests
diff --git a/src/sentry/testutils/pytest/relay.py b/src/sentry/testutils/pytest/relay.py
index d952a9c8d0c6..a97f84a58426 100644
--- a/src/sentry/testutils/pytest/relay.py
+++ b/src/sentry/testutils/pytest/relay.py
@@ -31,6 +31,22 @@ def _relay_server_container_name() -> str:
     return "sentry_test_relay_server"
 
 
+def _wait_until_ready(url: str, timeout: float) -> None:
+    deadline = time.monotonic() + timeout
+    last_result = "not probed"
+    while (remaining := deadline - time.monotonic()) > 0:
+        try:
+            status = requests.get(url, timeout=min(1, remaining)).status_code
+        except requests.RequestException as e:
+            last_result = f"{type(e).__name__}: {e}"
+        else:
+            if status == 200:
+                return
+            last_result = f"HTTP {status}"
+        time.sleep(max(0, min(0.05, deadline - time.monotonic())))
+    raise TimeoutError(last_result)
+
+
 def _get_template_dir():
     return path.abspath(path.join(path.dirname(__file__), "template"))
 
@@ -193,23 +209,17 @@ def relay_server(relay_server_setup, settings):
             # seconds, and exponential backoff overshoots that by up to the last sleep.
             # The deadline matches the previous backoff schedule's total.
             readiness_url = f"{url}/api/relay/healthcheck/ready/"
-            deadline = time.monotonic() + 25.5
-            while True:
+            try:
+                _wait_until_ready(readiness_url, timeout=25.5)
+            except TimeoutError as e:
                 try:
-                    if requests.get(readiness_url, timeout=1).status_code == 200:
-                        break
-                except Exception as ex:
-                    if time.monotonic() >= deadline:
-                        _log.exception(str(ex))
-                        raise ValueError(
-                            f"relay did not start in time (now: {datetime.datetime.now().isoformat()}) {url}:\n{container.logs().decode()}"
-                        ) from ex
-                else:
-                    if time.monotonic() >= deadline:
-                        raise ValueError(
-                            f"relay did not become ready in time (now: {datetime.datetime.now().isoformat()}) {readiness_url}:\n{container.logs().decode()}"
-                        )
-                time.sleep(0.05)
+                    logs = container.logs().decode()
+                finally:
+                    if not environ.get("RELAY_TEST_KEEP_CONTAINER", False):
+                        _remove_container_if_exists(docker_client, container_name)
+                raise ValueError(
+                    f"relay did not become ready in time (now: {datetime.datetime.now().isoformat()}, last result: {e}) {readiness_url}:\n{logs}"
+                ) from e
 
         yield {"url": relay_server_setup["url"]}
 
diff --git a/src/sentry/testutils/pytest/transactional_flush.py b/src/sentry/testutils/pytest/transactional_flush.py
index f554201e88ce..f19a11871510 100644
--- a/src/sentry/testutils/pytest/transactional_flush.py
+++ b/src/sentry/testutils/pytest/transactional_flush.py
@@ -15,6 +15,9 @@
 the last real flush produced, so ``post_migrate`` does not have to run again.
 When they don't match, Django's flush runs. Either way every table ends up
 empty apart from the ``post_migrate`` rows.
+
+The tables stay locked while the rows are counted and deleted, so a row
+committed in between can't outlive the test.
 """
 
 from __future__ import annotations
@@ -23,7 +26,7 @@
 
 import pytest
 from django.core.management import call_command
-from django.db import connections, transaction
+from django.db import OperationalError, connections, transaction
 from django.db.backends.base.base import BaseDatabaseWrapper
 from django.test import TransactionTestCase
 
@@ -56,25 +59,17 @@ def _tables_with_pages(connection: BaseDatabaseWrapper) -> frozenset[str]:
         return frozenset(row[0] for row in cursor.fetchall())
 
 
-def _other_transactions_open(connection: BaseDatabaseWrapper) -> bool:
+def _lock_tables(connection: BaseDatabaseWrapper) -> None:
     """
-    TRUNCATE waits for other open transactions (a leaked thread, say) to finish
-    and then removes their rows too. DELETE would not see their uncommitted
-    rows, so leave those cases to Django's flush.
+    Takes the same lock as TRUNCATE. DELETE can't see another transaction's
+    uncommitted rows, so if one has touched any table (a leaked thread, say)
+    this times out and Django's TRUNCATE waits for it instead.
     """
+    tables = connection.introspection.django_table_names(only_existing=True, include_views=False)
+    qn = connection.ops.quote_name
     with connection.cursor() as cursor:
-        cursor.execute(
-            """
-            SELECT EXISTS (
-                SELECT 1 FROM pg_stat_activity
-                WHERE datname = current_database()
-                  AND pid <> pg_backend_pid()
-                  AND xact_start IS NOT NULL
-            )
-            """
-        )
-        row = cursor.fetchone()
-        return bool(row and row[0])
+        cursor.execute("SET LOCAL lock_timeout = '1s'")
+        cursor.execute(f"LOCK TABLE {', '.join(qn(t) for t in tables)} IN ACCESS EXCLUSIVE MODE")
 
 
 def _fingerprint(connection: BaseDatabaseWrapper, tables: Collection[str]) -> str:
@@ -107,29 +102,35 @@ def _django_flush(db_name: str) -> None:
     _post_migrate_rows[db_name] = (seeded, _fingerprint(connection, seeded))
 
 
-def flush_database(db_name: str) -> None:
+def _delete_rows(db_name: str, seeded: tuple[frozenset[str], str]) -> bool:
+    """Delete every row post_migrate didn't create, or return False if that isn't safe."""
     connection = connections[db_name]
-    tables = _tables_with_pages(connection)
+    try:
+        with transaction.atomic(using=db_name):
+            _lock_tables(connection)
+            tables = _tables_with_pages(connection)
+            if not seeded[0] <= tables or _fingerprint(connection, seeded[0]) != seeded[1]:
+                return False
+            to_delete = sorted(tables - seeded[0])
+            if to_delete:
+                # Foreign keys are deferred, so the order of the deletes does
+                # not matter. The fence keeps the outbox write audit from
+                # flagging the deletes.
+                qn = connection.ops.quote_name
+                with unguarded_write(using=db_name), connection.cursor() as cursor:
+                    cursor.execute("; ".join(f"DELETE FROM {qn(t)}" for t in to_delete))
+    except OperationalError as e:
+        # lock_not_available or deadlock_detected
+        if getattr(e.__cause__, "pgcode", None) not in ("55P03", "40P01"):
+            raise
+        return False
+    return True
+
 
+def flush_database(db_name: str) -> None:
     seeded = _post_migrate_rows.get(db_name)
-    if (
-        seeded is None
-        or _other_transactions_open(connection)
-        or not seeded[0] <= tables
-        or _fingerprint(connection, seeded[0]) != seeded[1]
-    ):
+    if seeded is None or not _delete_rows(db_name, seeded):
         _django_flush(db_name)
-        return
-
-    to_delete = sorted(tables - seeded[0])
-    if not to_delete:
-        return
-    # Foreign keys are deferred, so the order of the deletes does not matter.
-    # The fence keeps the outbox write audit from flagging the deletes.
-    qn = connection.ops.quote_name
-    with transaction.atomic(using=db_name), unguarded_write(using=db_name):
-        with connection.cursor() as cursor:
-            cursor.execute("; ".join(f"DELETE FROM {qn(t)}" for t in to_delete))
 
 
 _original_fixture_teardown = TransactionTestCase._fixture_teardown  # type: ignore[attr-defined]
@@ -150,3 +151,8 @@ def _fixture_teardown(self: TransactionTestCase) -> None:
 
 def pytest_configure(config: pytest.Config) -> None:
     TransactionTestCase._fixture_teardown = _fixture_teardown  # type: ignore[attr-defined]
+
+
+def pytest_unconfigure(config: pytest.Config) -> None:
+    TransactionTestCase._fixture_teardown = _original_fixture_teardown  # type: ignore[attr-defined]
+    _post_migrate_rows.clear()
diff --git a/tests/sentry/spans/test_buffer.py b/tests/sentry/spans/test_buffer.py
index 2987263ebc79..21df2832d886 100644
--- a/tests/sentry/spans/test_buffer.py
+++ b/tests/sentry/spans/test_buffer.py
@@ -215,6 +215,12 @@ def process_spans(spans: Sequence[Span | _SplitBatch], buffer: SpansBuffer, now)
         buffer.process_spans(chunk, now)
 
 
+def _fresh_buffer(buffer: SpansBuffer) -> SpansBuffer:
+    # An earlier permutation can leave state such as `any_shard_at_limit` behind.
+    _delete_span_buffer_keys(buffer.client)
+    return SpansBuffer(assigned_shards=buffer.assigned_shards, slice_id=buffer.slice_id)
+
+
 def _span_order(spans: Sequence[Span | _SplitBatch]) -> str:
     """Name a permutation by span id initials, with `|` for a batch split, e.g. `d | b a c`."""
     return " ".join("|" if isinstance(span, _SplitBatch) else span.span_id[0] for span in spans)
@@ -570,7 +576,7 @@ def test_basic(buffer: SpansBuffer, subtests: pytest.Subtests) -> None:
         ]
     ):
         with subtests.test(msg=_span_order(spans)):
-            _delete_span_buffer_keys(buffer.client)
+            buffer = _fresh_buffer(buffer)
             process_spans(spans, buffer, now=0)
 
             assert_ttls(buffer.client)
@@ -767,7 +773,7 @@ def test_deep(buffer: SpansBuffer, subtests: pytest.Subtests) -> None:
     ]
     for spans in itertools.permutations(spans_to_permute):
         with subtests.test(msg=_span_order(spans)):
-            _delete_span_buffer_keys(buffer.client)
+            buffer = _fresh_buffer(buffer)
             process_spans(spans, buffer, now=0)
 
             assert_ttls(buffer.client)
@@ -843,7 +849,7 @@ def test_deep2(buffer: SpansBuffer, subtests: pytest.Subtests) -> None:
         ]
     ):
         with subtests.test(msg=_span_order(spans)):
-            _delete_span_buffer_keys(buffer.client)
+            buffer = _fresh_buffer(buffer)
             process_spans(spans, buffer, now=0)
 
             assert_ttls(buffer.client)
@@ -912,7 +918,7 @@ def test_parent_in_other_project(buffer: SpansBuffer, subtests: pytest.Subtests)
         ]
     ):
         with subtests.test(msg=_span_order(spans)):
-            _delete_span_buffer_keys(buffer.client)
+            buffer = _fresh_buffer(buffer)
             process_spans(spans, buffer, now=0)
 
             assert_ttls(buffer.client)
diff --git a/tests/sentry/testutils/pytest/test_json_report_reruns.py b/tests/sentry/testutils/pytest/test_json_report_reruns.py
new file mode 100644
index 000000000000..1a94982503f8
--- /dev/null
+++ b/tests/sentry/testutils/pytest/test_json_report_reruns.py
@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import textwrap
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from sentry.utils import json
+
+INNER_TESTS = textwrap.dedent(
+    """
+    attempts = 0
+
+    def test_subtests(subtests):
+        for i in range(3):
+            with subtests.test(msg="loop", i=i):
+                assert i != 1, "subtest boom"
+
+    def test_flaky_after_subtests(subtests):
+        global attempts
+        attempts += 1
+        with subtests.test(attempt=attempts):
+            pass
+        assert attempts > 1, "first attempt fails"
+    """
+)
+
+
+@pytest.fixture(params=[(), ("-n", "2", "--dist=loadfile")], ids=["inprocess", "xdist"])
+def tests(request: pytest.FixtureRequest, tmp_path: Path) -> dict[str, dict[str, Any]]:
+    (tmp_path / "pytest.ini").write_text("[pytest]\n")
+    (tmp_path / "test_inner.py").write_text(INNER_TESTS)
+    # Keep the outer run's options and settings out, and load exactly the
+    # plugins CI combines.
+    env = {
+        k: v for k, v in os.environ.items() if not k.startswith(("PYTEST_", "DJANGO_", "SENTRY_"))
+    }
+    env["PYTEST_DISABLE_PLUGIN_AUTOLOAD"] = "1"
+    plugins = (
+        "xdist.plugin",
+        "pytest_rerunfailures",
+        "pytest_jsonreport.plugin",
+        "sentry.testutils.pytest.json_report_reruns",
+    )
+    proc = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "pytest",
+            "-p",
+            "no:cacheprovider",
+            *(arg for plugin in plugins for arg in ("-p", plugin)),
+            "--json-report",
+            "--json-report-file=report.json",
+            "--reruns=1",
+            *request.param,
+            "test_inner.py",
+        ],
+        cwd=tmp_path,
+        env=env,
+        capture_output=True,
+        text=True,
+        timeout=120,
+    )
+    assert proc.returncode == 1, proc.stdout + proc.stderr
+    report = json.loads((tmp_path / "report.json").read_text())
+    return {test["nodeid"].split("::")[1]: test for test in report["tests"]}
+
+
+def test_subtests_are_not_reruns(tests: dict[str, dict[str, Any]]) -> None:
+    test = tests["test_subtests"]
+    assert "reruns" not in test
+    assert test["outcome"] == "failed"
+    assert [(s["outcome"], s["context"]) for s in test["subtests"]] == [
+        ("passed", {"msg": "loop", "kwargs": {"i": "0"}}),
+        ("failed", {"msg": "loop", "kwargs": {"i": "1"}}),
+        ("passed", {"msg": "loop", "kwargs": {"i": "2"}}),
+    ]
+    assert "AssertionError: subtest boom" in test["subtests"][1]["longrepr"]
+
+
+def test_rerun_keeps_its_own_subtests(tests: dict[str, dict[str, Any]]) -> None:
+    test = tests["test_flaky_after_subtests"]
+    [rerun] = test["reruns"]
+    assert rerun["call"]["outcome"] == "rerun"
+    assert [s["context"]["kwargs"] for s in rerun["subtests"]] == [{"attempt": "1"}]
+    assert [s["context"]["kwargs"] for s in test["subtests"]] == [{"attempt": "2"}]
+    assert test["call"]["outcome"] == "passed"
diff --git a/tests/sentry/testutils/pytest/test_relay.py b/tests/sentry/testutils/pytest/test_relay.py
new file mode 100644
index 000000000000..99a710f69b1c
--- /dev/null
+++ b/tests/sentry/testutils/pytest/test_relay.py
@@ -0,0 +1,50 @@
+from unittest import mock
+
+import pytest
+import requests
+
+from sentry.testutils.pytest import relay
+
+
+class FakeRelay:
+    def __init__(self, *answers: int | requests.RequestException) -> None:
+        self.answers = list(answers)
+        self.now = 0.0
+        self.probe_timeouts: list[float] = []
+
+    def get(self, url: str, timeout: float) -> mock.Mock:
+        self.probe_timeouts.append(timeout)
+        self.now += timeout
+        answer = self.answers.pop(0) if len(self.answers) > 1 else self.answers[0]
+        if isinstance(answer, Exception):
+            raise answer
+        return mock.Mock(status_code=answer)
+
+    def monotonic(self) -> float:
+        return self.now
+
+    def sleep(self, seconds: float) -> None:
+        assert seconds >= 0
+        self.now += seconds
+
+    def wait(self, timeout: float) -> None:
+        with (
+            mock.patch.object(requests, "get", self.get),
+            mock.patch.object(relay, "time", mock.Mock(monotonic=self.monotonic, sleep=self.sleep)),
+        ):
+            relay._wait_until_ready("http://relay/ready", timeout=timeout)
+
+
+def test_retries_until_ready() -> None:
+    fake = FakeRelay(requests.ConnectionError("refused"), 503, 200)
+    fake.wait(timeout=25.5)
+    assert len(fake.probe_timeouts) == 3
+
+
+def test_gives_up_at_deadline_with_last_result() -> None:
+    fake = FakeRelay(requests.Timeout("read timed out"))
+    with pytest.raises(TimeoutError, match="Timeout: read timed out"):
+        fake.wait(timeout=2.5)
+    # The last probe is capped by the time left, so the deadline isn't overshot.
+    assert fake.probe_timeouts == pytest.approx([1, 1, 0.4])
+    assert fake.now == pytest.approx(2.5)
diff --git a/tests/sentry/testutils/pytest/test_sentry.py b/tests/sentry/testutils/pytest/test_sentry.py
new file mode 100644
index 000000000000..f4550218a18e
--- /dev/null
+++ b/tests/sentry/testutils/pytest/test_sentry.py
@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import textwrap
+from collections import Counter
+from pathlib import Path
+
+import sentry
+from sentry.utils import json
+
+SCRIPTS = Path(sentry.__file__).parents[2] / ".github" / "workflows" / "scripts"
+SHARDS = 4
+
+# Runs pytest-selected.py the way CI does, with this project's durations
+# and no file split.
+DRIVER = textwrap.dedent(
+    """
+    import importlib, os, sys
+    from pathlib import Path
+    sys.path[0] = os.getcwd()
+    sys.path.append(os.environ["SCRIPTS"])
+    selected = importlib.import_module("pytest-selected")
+    selected.DURATIONS_FILE = Path("durations.json")
+    selected.PIECES_PER_SHARD = 1
+    raise SystemExit(selected.main(sys.argv[1:]))
+    """
+)
+
+MARKED = textwrap.dedent(
+    """
+    import pytest
+
+    @pytest.mark.snuba_ci
+    @pytest.mark.parametrize("n", range(10))
+    def test_marked(n):
+        pass
+
+    def test_unmarked():
+        pass
+    """
+)
+
+
+def _collect(project: Path, command: list[str], env: dict[str, str]) -> Counter[str]:
+    # Keep the outer run's sharding, plugins and settings out.
+    base = {
+        k: v
+        for k, v in os.environ.items()
+        if not k.startswith(("PYTEST_", "DJANGO_", "SENTRY_", "TEST_GROUP", "TOTAL_TEST_GROUPS"))
+        and k not in ("SELECTED_TESTS_FILE", "PYTHONPATH")
+    }
+    base |= {
+        "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1",
+        "SCRIPTS": str(SCRIPTS),
+        # How Snuba's CI filters sentry's tests.
+        "PYTEST_ADDOPTS": "-k 'not __In' -m snuba_ci",
+    }
+    proc = subprocess.run(
+        [*command, "-p", "no:cacheprovider", "tests", "--collect-only", "-q"],
+        cwd=project,
+        env=base | env,
+        capture_output=True,
+        text=True,
+        timeout=300,
+    )
+    # Exit code 5 is a shard that collected nothing.
+    assert proc.returncode == 0, proc.stdout + proc.stderr
+    return Counter(line for line in proc.stdout.splitlines() if "::" in line)
+
+
+def test_filtered_run_spreads_selected_tests_over_every_shard(tmp_path: Path) -> None:
+    files = {
+        "conftest.py": "from sentry.testutils.pytest.sentry import pytest_collection_modifyitems, pytest_ignore_collect  # noqa: F401\n",
+        "pytest.ini": "[pytest]\nmarkers =\n    snuba_ci\n",
+        "durations.json": json.dumps({"seconds_per_test": 0.1, "files": {}}),
+        **{f"tests/test_marked{i}.py": MARKED for i in range(2)},
+        **{f"tests/test_plain{i}.py": "def test_p():\n    pass\n" for i in range(12)},
+    }
+    for path, content in files.items():
+        (tmp_path / path).parent.mkdir(parents=True, exist_ok=True)
+        (tmp_path / path).write_text(content)
+
+    unsharded = _collect(tmp_path, [sys.executable, "-m", "pytest"], {})
+    sharded: Counter[str] = Counter()
+    for group in range(SHARDS):
+        sharded += _collect(
+            tmp_path,
+            [sys.executable, "-c", DRIVER],
+            {
+                "TOTAL_TEST_GROUPS": str(SHARDS),
+                "TEST_GROUP": str(group),
+                "TEST_GROUP_STRATEGY": "roundrobin",
+            },
+        )
+    assert len(unsharded) == 20
+    assert sharded == unsharded
diff --git a/tests/sentry/testutils/pytest/test_transactional_flush.py b/tests/sentry/testutils/pytest/test_transactional_flush.py
new file mode 100644
index 000000000000..ddc6dab27dca
--- /dev/null
+++ b/tests/sentry/testutils/pytest/test_transactional_flush.py
@@ -0,0 +1,190 @@
+from __future__ import annotations
+
+import os
+import threading
+import time
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
+from typing import Any
+from unittest import mock
+
+from django.contrib.auth.models import Permission
+from django.contrib.contenttypes.models import ContentType
+from django.contrib.sites.models import Site
+from django.db import connections
+from django.db.backends.base.base import BaseDatabaseWrapper
+
+from sentry.testutils.cases import TransactionTestCase
+from sentry.testutils.pytest import transactional_flush as tf
+from sentry.testutils.silo import no_silo_test
+
+ALIASES = ("default", "control", "secondary")
+WAIT_SECONDS = 30
+
+
+def _wait_for(condition: Callable[[], bool], what: str) -> None:
+    deadline = time.monotonic() + WAIT_SECONDS
+    while not condition():
+        assert time.monotonic() < deadline, f"timed out waiting for {what}"
+        time.sleep(0.01)
+
+
+def _scalar(sql: str, params: list[Any] | None = None) -> Any:
+    with connections["default"].cursor() as cursor:
+        cursor.execute(sql, params)
+        row = cursor.fetchone()
+        assert row is not None
+        return row[0]
+
+
+def _insert_option(key: str) -> None:
+    with connections["default"].cursor() as cursor:
+        cursor.execute(
+            "INSERT INTO sentry_option (key, value, last_updated, last_updated_by) "
+            "VALUES (%s, to_jsonb(%s::text), now(), 'unknown')",
+            [key, os.urandom(320 * 1024).hex()],
+        )
+
+
+def _tables_with_rows(alias: str) -> list[str]:
+    connection = connections[alias]
+    qn = connection.ops.quote_name
+    tables = connection.introspection.django_table_names(only_existing=True, include_views=False)
+    with connection.cursor() as cursor:
+        cursor.execute(
+            " UNION ALL ".join(f"SELECT %s WHERE EXISTS (SELECT 1 FROM {qn(t)})" for t in tables),
+            tables,
+        )
+        return sorted(row[0] for row in cursor.fetchall())
+
+
+@contextmanager
+def _count_django_flushes() -> Iterator[list[str]]:
+    flushed: list[str] = []
+    original = tf._django_flush
+
+    def spy(db_name: str) -> None:
+        flushed.append(db_name)
+        original(db_name)
+
+    with mock.patch.object(tf, "_django_flush", spy):
+        yield flushed
+
+
+@no_silo_test
+class TransactionalFlushTest(TransactionTestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        # Start from a recorded flush so no test depends on what ran before.
+        for alias in ALIASES:
+            tf.flush_database(alias)
+
+    def assert_flushed(self) -> None:
+        for alias in ALIASES:
+            seeded_tables, _ = tf._post_migrate_rows[alias]
+            assert _tables_with_rows(alias) == sorted(seeded_tables), alias
+
+    def test_deletes_rows_in_every_database(self) -> None:
+        org = self.create_organization(owner=self.create_user())
+        self.create_monitor(organization=org, project=self.create_project(organization=org))
+        _insert_option("raw-sql-and-toast")
+
+        with _count_django_flushes() as flushed:
+            for alias in ALIASES:
+                tf.flush_database(alias)
+
+        assert flushed == []
+        self.assert_flushed()
+
+    def test_changed_post_migrate_rows_fall_back_to_django_flush(self) -> None:
+        baseline = sorted(Permission.objects.values_list("codename", "name"))
+        Permission.objects.filter(pk=Permission.objects.order_by("pk")[0].pk).update(name="x")
+        ContentType.objects.create(app_label="flushtest", model="thing")
+        Site.objects.all().delete()
+
+        with _count_django_flushes() as flushed:
+            tf.flush_database("control")
+
+        assert flushed == ["control"]
+        assert sorted(Permission.objects.values_list("codename", "name")) == baseline
+        assert not ContentType.objects.filter(app_label="flushtest").exists()
+        assert Site.objects.exists()
+
+    def test_write_during_cleanup_is_deleted(self) -> None:
+        # An empty table has no pages, so it's left out of the inventory.
+        with connections["default"].cursor() as cursor:
+            cursor.execute("TRUNCATE sentry_option")
+        cleanup_pid = _scalar("SELECT pg_backend_pid()")
+        writer_pid: list[object] = []
+
+        def writer() -> None:
+            try:
+                writer_pid.append(_scalar("SELECT pg_backend_pid()"))
+                _insert_option("concurrent")
+            finally:
+                connections["default"].close()
+
+        thread = threading.Thread(target=writer)
+        original = tf._tables_with_pages
+
+        def inventory_then_write(connection: BaseDatabaseWrapper) -> frozenset[str]:
+            tables = original(connection)
+            if connection.alias == "default":
+                thread.start()
+                # Without the lock the insert would commit right here.
+                _wait_for(
+                    lambda: bool(writer_pid)
+                    and bool(
+                        _scalar(
+                            "SELECT %s = ANY(pg_blocking_pids(%s))", [cleanup_pid, writer_pid[0]]
+                        )
+                    ),
+                    "the writer to wait for cleanup",
+                )
+            return tables
+
+        with mock.patch.object(tf, "_tables_with_pages", inventory_then_write):
+            tf.flush_database("default")
+        thread.join(WAIT_SECONDS)
+
+        assert _scalar("SELECT count(*) FROM sentry_option") == 1
+        tf.flush_database("default")
+        assert _scalar("SELECT count(*) FROM sentry_option") == 0
+
+    def test_open_writer_falls_back_to_django_flush(self) -> None:
+        cleanup_pid = _scalar("SELECT pg_backend_pid()")
+        inserted = threading.Event()
+
+        def writer() -> None:
+            try:
+                with connections["default"].cursor() as cursor:
+                    cursor.execute("BEGIN")
+                    cursor.execute(
+                        "INSERT INTO sentry_option (key, value, last_updated, last_updated_by) "
+                        "VALUES ('open-writer', '1', now(), 'unknown')"
+                    )
+                    inserted.set()
+
+                    def truncate_waits() -> bool:
+                        cursor.execute("SELECT pg_stat_clear_snapshot()")
+                        cursor.execute(
+                            "SELECT query, wait_event_type FROM pg_stat_activity WHERE pid = %s",
+                            [cleanup_pid],
+                        )
+                        query, wait_event_type = cursor.fetchone()
+                        return query.startswith("TRUNCATE") and wait_event_type == "Lock"
+
+                    _wait_for(truncate_waits, "Django's flush to wait for the writer")
+                    cursor.execute("COMMIT")
+            finally:
+                connections["default"].close()
+
+        thread = threading.Thread(target=writer)
+        thread.start()
+        assert inserted.wait(WAIT_SECONDS)
+        with _count_django_flushes() as flushed:
+            tf.flush_database("default")
+        thread.join(WAIT_SECONDS)
+
+        assert flushed == ["default"]
+        assert _scalar("SELECT count(*) FROM sentry_option") == 0