diff --git a/changelog.d/20162.feature b/changelog.d/20162.feature
new file mode 100644
index 00000000000..a197d13664f
--- /dev/null
+++ b/changelog.d/20162.feature
@@ -0,0 +1 @@
+Limit the number of end-to-end encryption one-time keys stored per device to 500 per algorithm, rejecting uploads which would exceed the limit with a `400 Bad Request`.
diff --git a/synapse/handlers/e2e_keys.py b/synapse/handlers/e2e_keys.py
index 64f705a3dad..7c6e1c92134 100644
--- a/synapse/handlers/e2e_keys.py
+++ b/synapse/handlers/e2e_keys.py
@@ -20,6 +20,7 @@
 #
 #
 import logging
+from collections import Counter
 from typing import TYPE_CHECKING, Iterable, Mapping
 
 import attr
@@ -60,6 +61,18 @@
 
 ONE_TIME_KEY_UPLOAD = "one_time_key_upload_lock"
 
+# The maximum number of one-time keys, per algorithm, to keep for a device. Uploads
+# which would exceed it are rejected, which protects against clients that keep
+# uploading keys they will never be able to use.
+#
+# The spec allows clients to discard their oldest private one-time keys once they hold
+# too many, and both libolm (at 100 keys) and vodozemac (at 5000) do, so keys beyond
+# that bound could never be used anyway. The limit sits comfortably above the 50 keys
+# clients built on the matrix-rust-sdk aim to keep on the server, and well below
+# vodozemac's private key bound, so we never reject an upload from a well-behaved
+# client nor hold a key the client has already discarded.
+MAX_ONE_TIME_KEYS_PER_DEVICE = 500
+
 
 class E2eKeysHandler:
     def __init__(self, hs: "HomeServer"):
@@ -121,7 +134,6 @@ def __init__(self, hs: "HomeServer"):
         self._query_appservices_for_keys = (
             hs.config.experimental.msc3984_appservice_key_query
         )
-
         self._task_scheduler.register_action(
             self._delete_old_one_time_keys_task, "delete_old_otks"
         )
@@ -989,6 +1001,28 @@ async def _upload_one_time_keys_for_user(
                         (algorithm, key_id, encode_canonical_json(key).decode("ascii"))
                     )
 
+            # Reject uploads which would take the device over the limit, rather than
+            # quietly discarding keys, so that a client which keeps uploading keys
+            # regardless of how many the server holds gets told about it.
+            counts = await self.store.count_e2e_one_time_keys(user_id, device_id)
+            for algorithm, new_count in Counter(
+                algorithm for algorithm, _, _ in new_keys
+            ).items():
+                total = counts.get(algorithm, 0) + new_count
+                if total > MAX_ONE_TIME_KEYS_PER_DEVICE:
+                    raise SynapseError(
+                        400,
+                        "Uploading %i more %s one-time keys would leave the device "
+                        "holding %i, over the limit of %i"
+                        % (
+                            new_count,
+                            algorithm,
+                            total,
+                            MAX_ONE_TIME_KEYS_PER_DEVICE,
+                        ),
+                        Codes.TOO_LARGE,
+                    )
+
             log_kv({"message": "Inserting new one_time_keys.", "keys": new_keys})
             await self.store.add_e2e_one_time_keys(
                 user_id, device_id, time_now, new_keys
diff --git a/tests/handlers/test_e2e_keys.py b/tests/handlers/test_e2e_keys.py
index a4f9d55a133..b3b5953e6eb 100644
--- a/tests/handlers/test_e2e_keys.py
+++ b/tests/handlers/test_e2e_keys.py
@@ -405,6 +405,55 @@ def test_claim_one_time_key_bulk_ordering(self) -> None:
         for key_id in claimed_keys.keys():
             self.assertIn(key_id, ["alg1:k20", "alg1:k21", "alg1:k22"])
 
+    @mock.patch("synapse.handlers.e2e_keys.MAX_ONE_TIME_KEYS_PER_DEVICE", 5)
+    def test_upload_one_time_keys_over_limit_is_rejected(self) -> None:
+        """Uploading one-time keys which would take a device over the limit fails"""
+        local_user = "@boris:" + self.hs.hostname
+        device_id = "xyz"
+
+        # Uploading exactly up to the limit is fine.
+        res = self.get_success(
+            self.handler.upload_keys_for_user(
+                local_user,
+                device_id,
+                {"one_time_keys": {f"alg1:k{i}": f"key{i}" for i in range(1, 6)}},
+            )
+        )
+        self.assertEqual(res["one_time_key_counts"]["alg1"], 5)
+
+        # Uploading any more is rejected, with a 400.
+        error = self.get_failure(
+            self.handler.upload_keys_for_user(
+                local_user,
+                device_id,
+                {"one_time_keys": {"alg1:k6": "key6"}},
+            ),
+            SynapseError,
+        ).value
+        self.assertEqual(error.code, 400)
+        self.assertEqual(error.errcode, Codes.TOO_LARGE)
+
+        # The limit is per algorithm, so keys of another algorithm are still accepted.
+        res = self.get_success(
+            self.handler.upload_keys_for_user(
+                local_user,
+                device_id,
+                {"one_time_keys": {"alg2:k1": "key1"}},
+            )
+        )
+        self.assertEqual(res["one_time_key_counts"]["alg1"], 5)
+        self.assertEqual(res["one_time_key_counts"]["alg2"], 1)
+
+        # Nothing from the rejected upload was stored.
+        keys = self.get_success(
+            self.store.get_e2e_one_time_keys(
+                local_user, device_id, [f"k{i}" for i in range(1, 7)]
+            )
+        )
+        self.assertEqual(
+            set(keys), {("alg1", f"k{i}") for i in range(1, 6)} | {("alg2", "k1")}
+        )
+
     def test_fallback_key(self) -> None:
         local_user = "@boris:" + self.hs.hostname
         device_id = "xyz"