diff --git a/changelog.d/19832.misc b/changelog.d/19832.misc
new file mode 100644
index 0000000000..23c63fa10c
--- /dev/null
+++ b/changelog.d/19832.misc
@@ -0,0 +1 @@
+Fix a flaky test (`twisted.protocols.amp.TooLong` error under `trial -jN`) caused by an oversized debug log line.
diff --git a/synapse/storage/database.py b/synapse/storage/database.py
index 6e38b55686..39c0631d50 100644
--- a/synapse/storage/database.py
+++ b/synapse/storage/database.py
@@ -77,6 +77,13 @@ sql_logger = logging.getLogger("synapse.storage.SQL")
 transaction_logger = logging.getLogger("synapse.storage.txn")
 perf_logger = logging.getLogger("synapse.storage.TIME")
 
+# The maximum length of the repr of a query's values that we log at DEBUG. Some
+# queries carry large payloads (e.g. to-device messages), and logging them in
+# full is not useful. Truncating also keeps the line well under AMP's 64KiB
+# per-value limit, which would otherwise break `trial -jN` test runs (c.f.
+# https://github.com/twisted/twisted/issues/12482).
+MAX_SQL_VALUE_LOG_LENGTH = 1000
+
 sql_scheduling_timer = Histogram(
     "synapse_storage_schedule_time", "sec", labelnames=[SERVER_NAME_LABEL]
 )
@@ -502,7 +509,13 @@ class LoggingTransaction:
         sql = self.database_engine.convert_param_style(sql)
         if args:
             try:
-                sql_logger.debug("[SQL values] {%s} %r", self.name, args[0])
+                if sql_logger.isEnabledFor(logging.DEBUG):
+                    value_repr = repr(args[0])
+                    if len(value_repr) > MAX_SQL_VALUE_LOG_LENGTH:
+                        value_repr = (
+                            value_repr[:MAX_SQL_VALUE_LOG_LENGTH] + "... [truncated]"
+                        )
+                    sql_logger.debug("[SQL values] {%s} %s", self.name, value_repr)
             except Exception:
                 # Don't let logging failures stop SQL from working
                 pass
diff --git a/tests/logging/test_loggers.py b/tests/logging/test_loggers.py
index 9a9bf14376..815746b2dd 100644
--- a/tests/logging/test_loggers.py
+++ b/tests/logging/test_loggers.py
@@ -21,6 +21,19 @@ from tests.unittest import TestCase
 
 
 class ExplicitlyConfiguredLoggerTestCase(TestCase):
+    def setUp(self) -> None:
+        super().setUp()
+
+        # These tests mutate the root logger's level (see the various
+        # `root_logger.setLevel(...)` calls below). Make sure we restore the
+        # original level afterwards, otherwise we leak a raised log level into
+        # the other tests running in the same worker process. A raised level can
+        # cause oversized debug log lines which break `trial -jN` (worker log
+        # events are shipped to the manager over trials's AMP, which caps each
+        # value at 64KiB, c.f. https://github.com/twisted/twisted/issues/12482).
+        root_logger = logging.getLogger()
+        self.addCleanup(root_logger.setLevel, root_logger.level)
+
     def _create_explicitly_configured_logger(self) -> logging.Logger:
         original_logger_class = logging.getLoggerClass()
         logging.setLoggerClass(ExplicitlyConfiguredLogger)
diff --git a/tests/test_utils/logging_setup.py b/tests/test_utils/logging_setup.py
index d58222a9f6..bfae36436f 100644
--- a/tests/test_utils/logging_setup.py
+++ b/tests/test_utils/logging_setup.py
@@ -25,6 +25,14 @@ import twisted.logger
 from synapse.logging.context import LoggingContextFilter
 from synapse.synapse_rust import reset_logging_config
 
+# The maximum length of a single log line. Extremely long log lines aren't
+# useful, and keeping them short also stays well under the limit that would
+# otherwise break parallel test runs: under `trial -jN` worker log events are
+# shipped to the manager process over trial's AMP, which raises `TooLong` for
+# any value of 64KiB or more (e.g. a DEBUG dump of a large SQL query's values)
+# (c.f. https://github.com/twisted/twisted/issues/12482).
+_MAX_LOG_LINE_LENGTH = 60_000
+
 
 class ToTwistedHandler(logging.Handler):
     """logging handler which sends the logs to the twisted log"""
@@ -33,6 +41,11 @@ class ToTwistedHandler(logging.Handler):
 
     def emit(self, record: logging.LogRecord) -> None:
         log_entry = self.format(record)
+        if len(log_entry) > _MAX_LOG_LINE_LENGTH:
+            log_entry = (
+                log_entry[:_MAX_LOG_LINE_LENGTH]
+                + "... [truncated overly long log line]"
+            )
         log_level = record.levelname.lower().replace("warning", "warn")
         self.tx_log.emit(
             twisted.logger.LogLevel.levelWithName(log_level), "{entry}", entry=log_entry