diff --git a/changelog.d/19453.misc b/changelog.d/19453.misc
new file mode 100644
index 00000000000..48744c3a7a0
--- /dev/null
+++ b/changelog.d/19453.misc
@@ -0,0 +1 @@
+Introduce `spam_checker_spammy` internal event metadata.
\ No newline at end of file
diff --git a/rust/src/events/internal_metadata.rs b/rust/src/events/internal_metadata.rs
index fa40fdcfad8..38c6601d4f2 100644
--- a/rust/src/events/internal_metadata.rs
+++ b/rust/src/events/internal_metadata.rs
@@ -55,6 +55,7 @@ enum EventInternalMetadataData {
     SoftFailed(bool),
     ProactivelySend(bool),
     PolicyServerSpammy(bool),
+    SpamCheckerSpammy(bool),
     Redacted(bool),
     TxnId(Box<str>),
     TokenId(i64),
@@ -104,6 +105,13 @@ impl EventInternalMetadataData {
                     .to_owned()
                     .into_any(),
             ),
+            EventInternalMetadataData::SpamCheckerSpammy(o) => (
+                pyo3::intern!(py, "spam_checker_spammy"),
+                o.into_pyobject(py)
+                    .unwrap_infallible()
+                    .to_owned()
+                    .into_any(),
+            ),
             EventInternalMetadataData::Redacted(o) => (
                 pyo3::intern!(py, "redacted"),
                 o.into_pyobject(py)
@@ -168,6 +176,11 @@ impl EventInternalMetadataData {
                     .extract()
                     .with_context(|| format!("'{key_str}' has invalid type"))?,
             ),
+            "spam_checker_spammy" => EventInternalMetadataData::SpamCheckerSpammy(
+                value
+                    .extract()
+                    .with_context(|| format!("'{key_str}' has invalid type"))?,
+            ),
             "redacted" => EventInternalMetadataData::Redacted(
                 value
                     .extract()
@@ -451,6 +464,17 @@ impl EventInternalMetadata {
         set_property!(self, PolicyServerSpammy, obj);
     }
 
+    #[getter]
+    fn get_spam_checker_spammy(&self) -> PyResult<bool> {
+        Ok(get_property_opt!(self, SpamCheckerSpammy)
+            .copied()
+            .unwrap_or(false))
+    }
+    #[setter]
+    fn set_spam_checker_spammy(&mut self, obj: bool) {
+        set_property!(self, SpamCheckerSpammy, obj);
+    }
+
     #[getter]
     fn get_redacted(&self) -> PyResult<bool> {
         let bool = get_property!(self, Redacted)?;
diff --git a/synapse/federation/federation_base.py b/synapse/federation/federation_base.py
index 04ba5b86db0..fe0710a0bf4 100644
--- a/synapse/federation/federation_base.py
+++ b/synapse/federation/federation_base.py
@@ -177,7 +177,9 @@ async def _check_sigs_and_hash(
             # Note: we don't redact the event so admins can inspect the event after the
             # fact. Other processes may redact the event, but that won't be applied to
             # the database copy of the event until the server's config requires it.
-            return pdu
+            #
+            # We also *don't* return early here as we would still like to evaluate
+            # `spam_checker_spammy`, for completeness.
 
         spam_check = await self._spam_checker_module_callbacks.check_event_for_spam(pdu)
 
@@ -194,6 +196,8 @@ async def _check_sigs_and_hash(
             # using the event in prev_events).
             redacted_event = prune_event(pdu)
             redacted_event.internal_metadata.soft_failed = True
+            # Mark this as spam so we don't re-evaluate soft-failure status.
+            redacted_event.internal_metadata.spam_checker_spammy = True
             return redacted_event
 
         return pdu
diff --git a/synapse/storage/databases/main/sticky_events.py b/synapse/storage/databases/main/sticky_events.py
index 101306296e4..a2effcf2523 100644
--- a/synapse/storage/databases/main/sticky_events.py
+++ b/synapse/storage/databases/main/sticky_events.py
@@ -211,10 +211,14 @@ def insert_sticky_events_txn(
         Skips inserting events:
             - if they are considered spammy by the policy server;
               (unsure if correct, track: https://github.com/matrix-org/matrix-spec-proposals/pull/4354#discussion_r2727593350)
+            - if they are considered spammy by a Synapse spam checker module;
             - if they are rejected;
             - if they are outliers (they should be reconsidered for insertion when de-outliered); or
             - if they are not sticky (e.g. if the stickiness expired).
 
+        Note: Soft-failed sticky events ARE inserted, as their soft-failed status
+            could be re-evaluated later.
+
         Skipping the insertion of these types of 'invalid' events is useful for performance reasons because
         they would fill up the table yet we wouldn't show them to clients anyway.
 
@@ -230,7 +234,12 @@ def insert_sticky_events_txn(
         sticky_events: list[tuple[EventBase, int]] = []
         for ev in events:
             # MSC: Note: policy servers and other similar antispam techniques still apply to these events.
-            if ev.internal_metadata.policy_server_spammy:
+            # We don't filter out soft-failed events altogether (in case they get re-evaluated later),
+            # so filter out `spam_checker_spammy` events specifically as we don't want to re-evaluate _those_ later.
+            if (
+                ev.internal_metadata.policy_server_spammy
+                or ev.internal_metadata.spam_checker_spammy
+            ):
                 continue
             # We shouldn't be passed rejected events, but if we do, we filter them out too.
             if ev.rejected_reason is not None:
@@ -241,7 +250,7 @@ def insert_sticky_events_txn(
             sticky_duration = ev.sticky_duration()
             if sticky_duration is None:
                 continue
-            # Calculate the end time as start_time + effecitve sticky duration
+            # Calculate the end time as start_time + effective sticky duration
             expires_at = min(ev.origin_server_ts, now_ms) + sticky_duration.as_millis()
             # Filter out already expired sticky events
             if expires_at <= now_ms:
diff --git a/synapse/synapse_rust/events.pyi b/synapse/synapse_rust/events.pyi
index 0add391c651..9fcbe0f5d6d 100644
--- a/synapse/synapse_rust/events.pyi
+++ b/synapse/synapse_rust/events.pyi
@@ -36,6 +36,19 @@ class EventInternalMetadata:
     policy_server_spammy: bool
     """whether the policy server indicated that this event is spammy"""
 
+    spam_checker_spammy: bool
+    """Whether a spam checker module indicated that this event is spammy
+
+    Note that spam checkers also cause the event to be marked as soft-failed.
+
+    This flags exists for two reasons:
+        1. as debugging information
+        2. to prevent the soft-failed re-evaluation of spammy events
+           (the re-evaluation behaviour originates from MSC4354 Sticky Events)
+
+    Note that historical spammy events won't have this flag.
+    """
+
     txn_id: str
     """The transaction ID, if it was set when the event was created."""
     token_id: int
diff --git a/tests/module_api/test_spamchecker.py b/tests/module_api/test_spamchecker.py
index 42ef969ce01..572f7bdbcfa 100644
--- a/tests/module_api/test_spamchecker.py
+++ b/tests/module_api/test_spamchecker.py
@@ -12,22 +12,33 @@
 # <https://www.gnu.org/licenses/agpl-3.0.html>.
 #
 #
+from http import HTTPStatus
 from typing import Literal
 
 from twisted.internet.testing import MemoryReactor
 
-from synapse.api.constants import EventContentFields, EventTypes
+from synapse.api.constants import (
+    EventContentFields,
+    EventTypes,
+    Membership,
+)
+from synapse.api.room_versions import RoomVersions
 from synapse.config.server import DEFAULT_ROOM_VERSION
+from synapse.events import make_event_from_dict
+from synapse.module_api import EventBase
 from synapse.rest import admin, login, room, room_upgrade_rest_servlet
 from synapse.server import HomeServer
 from synapse.types import Codes, JsonDict
 from synapse.util.clock import Clock
 
+from tests import unittest
 from tests.server import FakeChannel
 from tests.unittest import HomeserverTestCase
 
 
 class SpamCheckerTestCase(HomeserverTestCase):
+    """Tests for the spam checker module API."""
+
     servlets = [
         room.register_servlets,
         admin.register_servlets,
@@ -284,3 +295,178 @@ async def user_may_send_state_event(
 
         self.assertEqual(channel.code, 403)
         self.assertEqual(channel.json_body["errcode"], Codes.FORBIDDEN)
+
+
+class FederatedEventSpamCheckMetadataTestCase(unittest.FederatingHomeserverTestCase):
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+    ]
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        super().prepare(reactor, clock, hs)
+        self._module_api = hs.get_module_api()
+        self._store = hs.get_datastores().main
+        self._storage_controllers = hs.get_storage_controllers()
+        self._federation_event_handler = hs.get_federation_event_handler()
+        self._federation_server = hs.get_federation_server()
+        self._state_handler = hs.get_state_handler()
+        self._persistence_controller = hs.get_storage_controllers().persistence
+
+        # Create a room
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        self.room_id = self.helper.create_room_as(
+            user1_id,
+            tok=user1_tok,
+            is_public=True,
+            room_version=RoomVersions.V10.identifier,
+        )
+
+        # Prepare a join for the 'remote' user
+        state_map = self.get_success(
+            self._storage_controllers.state.get_current_state(self.room_id)
+        )
+        forward_extremity_event_ids = self.get_success(
+            self.hs.get_datastores().main.get_latest_event_ids_in_room(self.room_id)
+        )
+        self.remote_user_id = f"@remoteuser:{self.OTHER_SERVER_NAME}"
+        self.remote_user_join_event = make_event_from_dict(
+            self.add_hashes_and_signatures_from_other_server(
+                {
+                    "room_id": self.room_id,
+                    "sender": self.remote_user_id,
+                    "state_key": self.remote_user_id,
+                    "depth": 1000,
+                    "origin_server_ts": 1,
+                    "type": EventTypes.Member,
+                    "content": {"membership": Membership.JOIN},
+                    "auth_events": [
+                        state_map[(EventTypes.Create, "")].event_id,
+                        state_map[(EventTypes.JoinRules, "")].event_id,
+                    ],
+                    "prev_events": list(forward_extremity_event_ids),
+                }
+            ),
+            room_version=RoomVersions.V10,
+        )
+
+        # Send the join
+        self.get_success(
+            self._federation_event_handler.on_receive_pdu(
+                self.OTHER_SERVER_NAME, self.remote_user_join_event
+            )
+        )
+
+        # Check the join made it to the 'local' view of the room
+        self.helper.get_event(
+            room_id=self.room_id,
+            event_id=self.remote_user_join_event.event_id,
+            tok=user1_tok,
+            expect_code=HTTPStatus.OK,
+        )
+
+    def test_federated_events_with_spam_checker_metadata(self) -> None:
+        """
+        Simulates receiving spammy and non-spammy events over federation,
+        then checks their `spam_checker_spammy` flag is set properly.
+        """
+
+        async def check_event_for_spam(event: EventBase) -> Literal["NOT_SPAM"] | Codes:
+            if event.type == EventTypes.Message:
+                if "ham" not in event.content["body"]:
+                    return Codes.FORBIDDEN
+            return "NOT_SPAM"
+
+        # Register a spam checker callback that only allows messages with 'ham'
+        self._module_api.register_spam_checker_callbacks(
+            check_event_for_spam=check_event_for_spam
+        )
+
+        # Prepare a spammy and a non-spammy event.
+        forward_extremity_event_ids = self.get_success(
+            self._store.get_latest_event_ids_in_room(self.room_id)
+        )
+        state_map = self.get_success(
+            self._storage_controllers.state.get_current_state(self.room_id)
+        )
+        spammy_event = make_event_from_dict(
+            self.add_hashes_and_signatures_from_other_server(
+                {
+                    "room_id": self.room_id,
+                    "sender": self.remote_user_id,
+                    "depth": 2000,
+                    "origin_server_ts": 2,
+                    "type": EventTypes.Message,
+                    "content": {"body": "this is spam", "msgtype": "m.text"},
+                    "auth_events": [
+                        state_map[(EventTypes.Create, "")].event_id,
+                        state_map[(EventTypes.JoinRules, "")].event_id,
+                        state_map[(EventTypes.Member, self.remote_user_id)].event_id,
+                    ],
+                    "prev_events": list(forward_extremity_event_ids),
+                }
+            ),
+            room_version=RoomVersions.V10,
+        )
+        non_spammy_event = make_event_from_dict(
+            self.add_hashes_and_signatures_from_other_server(
+                {
+                    "room_id": self.room_id,
+                    "sender": self.remote_user_id,
+                    "depth": 2000,
+                    "origin_server_ts": 2,
+                    "type": EventTypes.Message,
+                    "content": {"body": "delicious ham", "msgtype": "m.text"},
+                    "auth_events": [
+                        state_map[(EventTypes.Create, "")].event_id,
+                        state_map[(EventTypes.JoinRules, "")].event_id,
+                        state_map[(EventTypes.Member, self.remote_user_id)].event_id,
+                    ],
+                    "prev_events": list(forward_extremity_event_ids),
+                }
+            ),
+            room_version=RoomVersions.V10,
+        )
+
+        # Receive these events over federation
+        # We need to let the federation server have them because it will
+        # invoke `_check_sigs_and_hash` which invokes the spam checker.
+        self.get_success(
+            self._federation_server._handle_received_pdu(
+                self.OTHER_SERVER_NAME, spammy_event
+            )
+        )
+        self.get_success(
+            self._federation_server._handle_received_pdu(
+                self.OTHER_SERVER_NAME, non_spammy_event
+            )
+        )
+
+        # Retrieve the events from the database
+        retrieved_spammy_event = self.get_success(
+            self._store.get_event(spammy_event.event_id, allow_rejected=True)
+        )
+        retrieved_non_spammy_event = self.get_success(
+            self._store.get_event(non_spammy_event.event_id, allow_rejected=True)
+        )
+
+        # Assert the spammy flags (and soft-failed flags, for good measure) are set properly
+        self.assertTrue(
+            retrieved_spammy_event.internal_metadata.spam_checker_spammy,
+            "Spammy inbound event should be marked as spam_checker_spammy!",
+        )
+        self.assertTrue(
+            retrieved_spammy_event.internal_metadata.is_soft_failed(),
+            "Spammy inbound event should be soft-failed.",
+        )
+
+        self.assertFalse(
+            retrieved_non_spammy_event.internal_metadata.spam_checker_spammy,
+            "Non-spammy inbound event should not be marked as spam_checker_spammy!",
+        )
+        self.assertFalse(
+            retrieved_non_spammy_event.internal_metadata.is_soft_failed(),
+            "Non-spammy inbound event should not be soft-failed.",
+        )
diff --git a/tests/rest/client/sliding_sync/test_rooms_meta.py b/tests/rest/client/sliding_sync/test_rooms_meta.py
index 9e3f8aaf943..4ff07b4d261 100644
--- a/tests/rest/client/sliding_sync/test_rooms_meta.py
+++ b/tests/rest/client/sliding_sync/test_rooms_meta.py
@@ -12,6 +12,7 @@
 # <https://www.gnu.org/licenses/agpl-3.0.html>.
 #
 import logging
+from typing import Any
 
 from parameterized import parameterized, parameterized_class
 
@@ -966,7 +967,7 @@ def test_rooms_bump_stamp_backfill(self) -> None:
         creator = "@user:other"
         room_id = "!foo:other"
         room_version = RoomVersions.V10
-        shared_kwargs = {
+        shared_kwargs: dict[str, Any] = {
             "room_id": room_id,
             "room_version": room_version.identifier,
         }
diff --git a/tests/storage/test_roommember.py b/tests/storage/test_roommember.py
index f8d64e8ce68..711c9ba4ac6 100644
--- a/tests/storage/test_roommember.py
+++ b/tests/storage/test_roommember.py
@@ -20,7 +20,7 @@
 #
 #
 import logging
-from typing import cast
+from typing import Any, cast
 
 from twisted.internet.testing import MemoryReactor
 
@@ -238,7 +238,7 @@ def test_join_locally_forgotten_room(self) -> None:
         creator = "@user:other"
         room_id = "!foo:other"
         room_version = RoomVersions.V10
-        shared_kwargs = {
+        shared_kwargs: dict[str, Any] = {
             "room_id": room_id,
             "room_version": room_version.identifier,
         }
diff --git a/tests/storage/test_sliding_sync_tables.py b/tests/storage/test_sliding_sync_tables.py
index f5bbd496631..39226ff9bef 100644
--- a/tests/storage/test_sliding_sync_tables.py
+++ b/tests/storage/test_sliding_sync_tables.py
@@ -18,7 +18,7 @@
 #
 #
 import logging
-from typing import cast
+from typing import Any, cast
 
 import attr
 from parameterized import parameterized
@@ -873,7 +873,7 @@ def test_joined_room_bump_stamp_backfill(self) -> None:
         creator = "@user:other"
         room_id = "!foo:other"
         room_version = RoomVersions.V10
-        shared_kwargs = {
+        shared_kwargs: dict[str, Any] = {
             "room_id": room_id,
             "room_version": room_version.identifier,
         }
diff --git a/tests/storage/test_sticky_events.py b/tests/storage/test_sticky_events.py
index 60243cb2f40..e77b362f528 100644
--- a/tests/storage/test_sticky_events.py
+++ b/tests/storage/test_sticky_events.py
@@ -30,6 +30,7 @@
 from synapse.util.duration import Duration
 
 from tests import unittest
+from tests.test_utils.event_injection import inject_event
 from tests.utils import USE_POSTGRES_FOR_TESTS
 
 
@@ -276,3 +277,154 @@ def test_outlier_events_not_in_table(self) -> None:
         )
         self.assertEqual(len(updates), 1)
         self.assertEqual(updates[0].event_id, event_non_outlier.event_id)
+
+    def test_soft_failed_events_are_tracked(self) -> None:
+        """
+        Tests that sticky events marked as soft_failed ARE inserted
+        into the sticky_events table, as their soft-failed status can be re-evaluated later,
+        as per MSC4354.
+        """
+        user_id = self.register_user("testuser", "pass")
+        token = self.login(user_id, "pass")
+        room_id = self.helper.create_room_as(user_id, tok=token)
+
+        start_id = self.store.get_max_sticky_events_stream_id()
+
+        # Create and persist a sticky event that is soft-failed
+        soft_failed_sticky_event = self.get_success(
+            inject_event(
+                self.hs,
+                room_id=room_id,
+                sender=user_id,
+                type=EventTypes.Message,
+                content={"body": "spam checker spammy message", "msgtype": "m.text"},
+                internal_metadata={"soft_failed": True},
+                # Corresponds to StickyEvent.EVENT_FIELD_NAME
+                msc4354_sticky=StickyEventField(
+                    duration_ms=Duration(minutes=1).as_millis()
+                ),
+            )
+        )
+
+        end_id = self.store.get_max_sticky_events_stream_id()
+
+        updates = self.get_success(
+            self.store.get_updated_sticky_events(
+                from_id=start_id, to_id=end_id, limit=10
+            )
+        )
+
+        self.assertEqual(len(updates), 1)
+        self.assertEqual(updates[0].event_id, soft_failed_sticky_event.event_id)
+
+    def test_policy_server_spammy_events_are_not_tracked(self) -> None:
+        """
+        Tests that sticky events marked as policy_server_spammy are NOT inserted
+        into the sticky_events table, as they are exempt from the soft-failed
+        re-evaluation logic.
+        """
+        user_id = self.register_user("testuser", "pass")
+        token = self.login(user_id, "pass")
+        room_id = self.helper.create_room_as(user_id, tok=token)
+
+        start_id = self.store.get_max_sticky_events_stream_id()
+
+        # Create and persist a sticky event that is marked policy_server_spammy
+        # N.B. policy_server_spammy events are always soft-failed too
+        _spammy_sticky_event = self.get_success(
+            inject_event(
+                self.hs,
+                room_id=room_id,
+                sender=user_id,
+                type=EventTypes.Message,
+                content={"body": "spam checker spammy message", "msgtype": "m.text"},
+                internal_metadata={"soft_failed": True, "policy_server_spammy": True},
+                # Corresponds to StickyEvent.EVENT_FIELD_NAME
+                msc4354_sticky=StickyEventField(
+                    duration_ms=Duration(minutes=1).as_millis()
+                ),
+            )
+        )
+
+        # Also insert a valid sticky event as a canary for the test setup
+        valid_sticky_event = self.get_success(
+            inject_event(
+                self.hs,
+                room_id=room_id,
+                sender=user_id,
+                type=EventTypes.Message,
+                content={"body": "normal sticky", "msgtype": "m.text"},
+                # Corresponds to StickyEvent.EVENT_FIELD_NAME
+                msc4354_sticky=StickyEventField(
+                    duration_ms=Duration(minutes=1).as_millis()
+                ),
+            )
+        )
+
+        end_id = self.store.get_max_sticky_events_stream_id()
+
+        # Verify only the regular event was inserted
+        updates = self.get_success(
+            self.store.get_updated_sticky_events(
+                from_id=start_id, to_id=end_id, limit=10
+            )
+        )
+
+        self.assertEqual(len(updates), 1)
+        self.assertEqual(updates[0].event_id, valid_sticky_event.event_id)
+
+    def test_spam_checker_spammy_events_are_not_tracked(self) -> None:
+        """
+        Tests that sticky events marked as spam_checker_spammy are NOT inserted
+        into the sticky_events table, as they are exempt from the soft-failed
+        re-evaluation logic.
+        """
+        user_id = self.register_user("testuser", "pass")
+        token = self.login(user_id, "pass")
+        room_id = self.helper.create_room_as(user_id, tok=token)
+
+        start_id = self.store.get_max_sticky_events_stream_id()
+
+        # Create and persist a sticky event that is marked spam_checker_spammy
+        # N.B. spam_checker_spammy events are always soft-failed too
+        _spammy_sticky_event = self.get_success(
+            inject_event(
+                self.hs,
+                room_id=room_id,
+                sender=user_id,
+                type=EventTypes.Message,
+                content={"body": "spam checker spammy message", "msgtype": "m.text"},
+                internal_metadata={"soft_failed": True, "spam_checker_spammy": True},
+                # Corresponds to StickyEvent.EVENT_FIELD_NAME
+                msc4354_sticky=StickyEventField(
+                    duration_ms=Duration(minutes=1).as_millis()
+                ),
+            )
+        )
+
+        # Also insert a valid sticky event as a canary for the test setup
+        valid_sticky_event = self.get_success(
+            inject_event(
+                self.hs,
+                room_id=room_id,
+                sender=user_id,
+                type=EventTypes.Message,
+                content={"body": "normal sticky", "msgtype": "m.text"},
+                # Corresponds to StickyEvent.EVENT_FIELD_NAME
+                msc4354_sticky=StickyEventField(
+                    duration_ms=Duration(minutes=1).as_millis()
+                ),
+            )
+        )
+
+        end_id = self.store.get_max_sticky_events_stream_id()
+
+        # Verify only the valid sticky event was inserted
+        updates = self.get_success(
+            self.store.get_updated_sticky_events(
+                from_id=start_id, to_id=end_id, limit=10
+            )
+        )
+
+        self.assertEqual(len(updates), 1)
+        self.assertEqual(updates[0].event_id, valid_sticky_event.event_id)
diff --git a/tests/test_utils/event_injection.py b/tests/test_utils/event_injection.py
index a90fc5884d8..36f6baa1239 100644
--- a/tests/test_utils/event_injection.py
+++ b/tests/test_utils/event_injection.py
@@ -18,7 +18,7 @@
 # [This file includes modifications made by New Vector Limited]
 #
 #
-from typing import Any
+from typing import Any, Mapping
 
 import synapse.server
 from synapse.api.constants import EventTypes
@@ -63,6 +63,8 @@ async def inject_event(
     hs: synapse.server.HomeServer,
     room_version: str | None = None,
     prev_event_ids: list[str] | None = None,
+    *,
+    internal_metadata: Mapping[str, Any] | None = None,
     **kwargs: Any,
 ) -> EventBase:
     """Inject a generic event into a room
@@ -72,9 +74,12 @@ async def inject_event(
         room_version: the version of the room we're inserting into.
             if not specified, will be looked up
         prev_event_ids: prev_events for the event. If not specified, will be looked up
+        internal_metadata: Dict representing the event's internal metadata; see `EventBase.internal_metadata`
         kwargs: fields for the event to be created
     """
-    event, context = await create_event(hs, room_version, prev_event_ids, **kwargs)
+    event, context = await create_event(
+        hs, room_version, prev_event_ids, internal_metadata=internal_metadata, **kwargs
+    )
 
     persistence = hs.get_storage_controllers().persistence
     assert persistence is not None
@@ -88,8 +93,11 @@ async def create_event(
     hs: synapse.server.HomeServer,
     room_version: str | None = None,
     prev_event_ids: list[str] | None = None,
+    *,
+    internal_metadata: Mapping[str, Any] | None = None,
     **kwargs: Any,
 ) -> tuple[EventBase, EventContext]:
+    internal_metadata = internal_metadata or {}
     if room_version is None:
         room_version = await hs.get_datastores().main.get_room_version_id(
             kwargs["room_id"]
@@ -106,11 +114,13 @@ async def create_event(
     )
 
     # Copy over writable internal_metadata, if set
-    # Dev note: This isn't everything that's writable. `for k,v` doesn't work here :(
-    if kwargs.get("internal_metadata", {}).get("soft_failed", False):
-        event.internal_metadata.soft_failed = True
-    if kwargs.get("internal_metadata", {}).get("policy_server_spammy", False):
-        event.internal_metadata.policy_server_spammy = True
+    if internal_metadata:
+        for key, value in internal_metadata.items():
+            # Note: this calls the relevant `#[setter]` function in the (Rust) event class'
+            # internal metadata struct.
+            # Will reject unknown keys with exceptions.
+            # This is desirable for our test suite anyway.
+            setattr(event.internal_metadata, key, value)
 
     context = await unpersisted_context.persist(event)