adapter: fix priority inversion that pins the coordinator on linearize_reads

[antiguru]

message_linearize_reads re-checks strict serializable reads whose chosen timestamp is ahead of the oracle by re-arming on the high-priority internal command channel, which the biased coordinator select polls above the group commit that advances the oracle.

Under sustained strict serializable load this is a self-sustaining priority inversion: the re-arm starves group commit, the oracle read timestamp freezes, the reads never become ready, and the re-arm spins forever. Group commit stops, the read-timestamp SELECT saturates the oracle, and the coordinator is reported stuck on linearize_reads until clients disconnect and drain the pending reads.

Signal the re-check via a dedicated tokio::sync::Notify that serve awaits below the group commit branches, so group commit always advances the oracle and the waiting reads retire instead of spinning. Re-check timing is otherwise unchanged; pacing it down is left to #37317.

No automated test: this is a scheduling/priority bug that's hard to unit-test. Observable signals that it stays fixed are the absence of the coordinator stuck for … last_message_kind: linearize_reads warning, and the requeued (false) bucket of the linearize_message_seconds histogram staying low (reads linearize on first check).

🤖 Generated with Claude Code

[aljoscha]

Change looks good! One thing we might want to tighten up is comments. I get that this is subtle, but we're basically repeating the same description and reasoning in all the new code sites.

[antiguru]

Thanks! Consolidated the reasoning into the serve select branch (where the ordering is visible) and trimmed the struct-field doc and the message_linearize_reads re-arm to one-line pointers. The pin-site comment stays since it explains the distinct cancel-safety/lost-wakeup concern. Pushed as 17c6a08.

[antiguru]

Thanks for the review

[def-]

This and the previous block still starve on traffic. We could reorder to fix that probably? Here is an extra parallel-benchmark scenario:

diff --git a/misc/python/materialize/parallel_benchmark/scenarios.py b/misc/python/materialize/parallel_benchmark/scenarios.py
index 06823e6fb1..0657c3b6be 100644
--- a/misc/python/materialize/parallel_benchmark/scenarios.py
+++ b/misc/python/materialize/parallel_benchmark/scenarios.py
@@ -258,6 +258,135 @@ class PgReadReplicaRTR(Scenario):
         )


+class LinearizeReadStarvation(Scenario):
+    """
+    Regression test for the residual coordinator priority inversion left by
+    PR #37316 (linearize-read starvation under sustained client traffic).
+
+    #37316 moved the re-check of pending strict serializable reads off the
+    high-priority internal command channel onto a dedicated `Notify` that
+    `Coordinator::serve` awaits below the group commit branches. That stops the
+    re-check from starving group commit, but the `Notify` branch, and the
+    strict-serializable-read receive branch that first enqueues a pending read,
+    both still sit below the `cmd_rx` user-command branch in the biased select.
+
+    Real-time recency pushes a strict serializable read to a timestamp ahead of
+    the oracle, so it becomes a pending linearize read. Retiring it needs one of
+    those lower-priority branches to run. Under a sustained write flood the
+    `cmd_rx` branch is runnable on every poll and starves the retirement, even
+    though the same flood keeps advancing the oracle past the read's chosen
+    timestamp on the high-priority group commit branch. The read stays pending
+    until the flood drains.
+
+    The scenario drives both halves at once:
+
+    * A heavy closed-loop `INSERT` flood into a native table. Its group commits
+      advance the EpochMilliseconds oracle on the high-priority branch, and the
+      insert commands keep `cmd_rx` non-empty well past `MESSAGE_BATCH` (64).
+    * Real-time-recency strict serializable reads over a Postgres-source
+      materialized view. Real-time recency pushes them ahead of the oracle, so
+      they become pending linearize reads that the flood then starves.
+
+    On current `main` the reads are starved and only retire once the flood stops
+    at the end of the load phase, so their recorded p99 latency approaches the
+    load-phase duration and the guarantee below fails. Retiring pending reads
+    from `Message::AdvanceTimelines` (delivered on the highest-priority internal
+    channel), or reordering the select so timeline advancement and the re-check
+    cannot be starved by user command intake, makes the reads retire promptly
+    and the guarantee pass.
+    """
+
+    def __init__(self, c: Composition, conn_infos: dict[str, PgConnInfo]):
+        self.init(
+            [
+                TdPhase("""
+                    > DROP SECRET IF EXISTS pgpass CASCADE
+                    > CREATE SECRET pgpass AS 'postgres'
+                    > CREATE CONNECTION pg TO POSTGRES (
+                        HOST postgres,
+                        DATABASE postgres,
+                        USER postgres,
+                        PASSWORD SECRET pgpass
+                      )
+
+                    $ postgres-execute connection=postgres://postgres:postgres@postgres
+                    DROP PUBLICATION IF EXISTS mz_linearize_starvation;
+                    DROP TABLE IF EXISTS lin_src CASCADE;
+                    ALTER USER postgres WITH replication;
+                    CREATE TABLE lin_src (f1 INTEGER);
+                    ALTER TABLE lin_src REPLICA IDENTITY FULL;
+                    CREATE PUBLICATION mz_linearize_starvation FOR ALL TABLES;
+
+                    > CREATE SOURCE lin_source
+                      FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_linearize_starvation')
+
+                    > CREATE TABLE lin_src FROM SOURCE lin_source (REFERENCE lin_src)
+
+                    > CREATE MATERIALIZED VIEW lin_mv AS
+                      SELECT COUNT(*) FROM lin_src
+
+                    > CREATE DEFAULT INDEX ON lin_mv
+
+                    # Native table for the coordinator command flood. Writing to it
+                    # advances the same EpochMilliseconds oracle the reads wait on.
+                    > CREATE TABLE lin_flood (f1 INTEGER)
+                    """),
+                LoadPhase(
+                    duration=120,
+                    actions=[
+                        # Keep the Postgres source frontier advancing so real-time
+                        # recency keeps pushing the reads ahead of the oracle.
+                        OpenLoop(
+                            action=StandaloneQuery(
+                                "INSERT INTO lin_src VALUES (1)",
+                                conn_infos["postgres"],
+                            ),
+                            dist=Periodic(per_second=100),
+                            report_regressions=False,
+                        ),
+                    ]
+                    + [
+                        # Sustained write flood. Its group commits advance the oracle
+                        # on the high-priority branch, and its commands keep cmd_rx
+                        # saturated (128 concurrent writers, well past MESSAGE_BATCH).
+                        ClosedLoop(
+                            action=ReuseConnQuery(
+                                "INSERT INTO lin_flood VALUES (1)",
+                                conn_infos["materialized"],
+                                strict_serializable=False,
+                            ),
+                            report_regressions=False,
+                        )
+                        for _ in range(1024)
+                    ]
+                    + [
+                        # Real-time-recency strict serializable reads. These become
+                        # pending linearize reads whose retirement the flood starves.
+                        ClosedLoop(
+                            action=ReuseConnQuery(
+                                "SET REAL_TIME_RECENCY TO TRUE; SELECT * FROM lin_mv",
+                                conn_infos["materialized"],
+                                strict_serializable=True,
+                            ),
+                            report_regressions=False,
+                        )
+                        for _ in range(8)
+                    ],
+                ),
+            ],
+            guarantees={
+                # RED against current main: the starved reads only retire when the
+                # flood drains at the end of the load phase, so p99 approaches the
+                # load-phase duration. Passes once pending reads retire from
+                # Message::AdvanceTimelines or the select is reordered so the
+                # re-check is not starved by cmd_rx.
+                "SET REAL_TIME_RECENCY TO TRUE; SELECT * FROM lin_mv (reuse connection)": {
+                    "p99": 5000,
+                },
+            },
+        )
+
+
 class MySQLReadReplica(Scenario):
     def __init__(self, c: Composition, conn_infos: dict[str, PgConnInfo]):
         self.init(

Running bin/mzcompose --find parallel-benchmark run default --scenario LinearizeReadStarvation --load-phase-duration 60:

Statistics for SET REAL_TIME_RECENCY TO TRUE; SELECT * FROM lin_mv (reuse connection):
  queries:     8
  qps:    5.51
  min: 60157.58ms
  avg: 60158.59ms
  p50: 60158.04ms
  p95: 60161.19ms
  p99: 60162.34ms
  max: 60162.62ms
  std:    1.66ms
  slope: -1943.4671

So all real-time-recency queries are stuck for the entire load period: