perf: batch pg_notify round trips in the runs-service NOTIFY pump #7756

Why

The runs-service sends one PostgreSQL NOTIFY per action update, and each one costs a full
network round trip on a single dedicated connection. That serial round trip — not Postgres, not
CPU — is what limits action-update throughput today.

Measured on a dev cluster during a 100,000-action benchmark:

The mean is 780 ms, so this is the whole distribution being slow, not a tail. And it is not
the usual suspects:

• Not Postgres — RDS WriteLatency 0.9 ms, CPU 58–72%, and only 14 database connections
  open against a pool max of 100. Every connection sat in ClientRead, i.e. the database was
  waiting on the application.
• Not CPU — the pod used 1.34 of its 4 cores with CFS throttling at 0.09% of periods.

By Little's law, 101.8 rps × 0.78 s ≈ 78 requests in flight — with an idle CPU and an idle
connection pool. Those requests are blocked on something inside the process.

They are blocked on the NOTIFY pump. pg_stat_activity shows the single dedicated connection:

 pid  | state | wait_event | query
 7330 | idle  | ClientRead | SELECT pg_notify($1, $2)     <- the pump, one connection

Timing pg_notify from a pod in the same VPC shows where the cost actually is:

pg_notify itself costs 32 µs; the other ~4.07 ms is pure round trip. Since the pump pays that
per payload, serially, its ceiling is ~240 notifications/second on an idle database — and it
degrades as the database gets busier. Demand at the measured rates (UpdateActionStatus alone at
101.8/s, plus RecordAction at 52.9/s, plus six other call sites) lands at or above that ceiling.

The user-visible effect: a swarm of 50 concurrent runs × 2,000 tasks (100k actions) finished only
37/50 runs before timing out at 6,014 s, while the executor sat at 63% of its memory limit with
zero restarts. The deployment is throughput-bound here, not memory-bound.

What to change

All in runs/repository/impl/action.go.

drainAndExec already drains the channel into a batch — it just issues one
execNotify round trip per payload instead of one per batch:

drainAndExec := func(channel, firstPayload string, ch <-chan string) {
    execNotify(channel, firstPayload)        // <- one round trip
    for {
        select {
        case payload, ok := <-ch:
            if !ok { return }
            execNotify(channel, payload)     // <- one round trip each
        default:
            return
        }
    }
}

Send the whole drained batch in one multi-call statement:

SELECT pg_notify($1,$2), pg_notify($1,$3), pg_notify($1,$4), ...

The channel name is the same for every call in a batch, so it binds once as $1 and each payload
adds one parameter. This keeps one notification per payload, which means the listener side is
completely untouched — no payload framing, no splitting, no changes to processNotifications or to any of the Watch* consumers.

Two constraints to respect when building the statement:

• Bind-parameter limit. PostgreSQL allows at most 65,535 parameters per statement, so a batch
  of N payloads uses N+1. Chunk well below that — a cap in the low thousands keeps the statement
  string small and parse time negligible. (InsertEvents already chunks for the same reason and is a good model.)
• Payload size is per notification, not per batch — the 8000-byte NOTIFY limit applies to each
  individual payload, and these payloads are short identifiers, so it is not a concern here.

A useful property worth knowing: PostgreSQL documents that when the same channel is signalled
multiple times with identical payloads inside one transaction, only one notification is
delivered. Since a batched statement is a single implicit transaction, duplicate payloads within a
batch collapse for free — the pump gets a degree of coalescing without writing any.

Retry the batch — don't widen the existing drop

Batching changes the blast radius of a failure, so it has to come with a retry.
execNotify currently discards a payload on any error:

if conn == nil {
    logger.Errorf(ctx, "No NOTIFY connection available, dropping %s notification", channel)
    return                                    // payload lost
}
if _, err := conn.ExecContext(ctx, "SELECT pg_notify($1, $2)", channel, payload); err != nil {
    logger.Errorf(ctx, "Failed to NOTIFY %s: %v", channel, err)
    if isConnError(err) { reconnect() }
    // payload lost — no retry
}

Today a connection blip loses one notification. Batch 500 payloads into one statement and the
same blip loses all 500 — 500 runs stuck at a stale phase in the UI until something else
touches them. That is a real regression if batching lands on its own.

So: hold the drained payloads in a slice until the statement succeeds. On failure, reconnect and
retry the same batch with a backoff instead of dropping it. Duplicate delivery is harmless — a
notification is an idempotent wakeup that causes a re-read — so at-least-once is the right target
and worth far more than avoiding a redundant wakeup.

If #7757 (the coalescing pending set) lands first, this becomes even simpler: merge the failed
batch's keys back into the pending set, which is idempotent by construction. The two issues
converge on the same retry story, so whoever goes second should reuse what the first built rather
than adding a parallel mechanism.

Files:

• runs/repository/impl/action.go — drainAndExec / execNotify
  inside runNotifyLoop.
• runs/repository/impl/action_test.go — add coverage that N queued updates produce N logical
  notifications with far fewer round trips.

The listener side must keep working unchanged: WatchRunUpdates,
WatchAllRunUpdates, WatchAllActionUpdates and
WatchActionUpdates all consume these notifications.

Outcome

• A drained batch of N payloads costs one database round trip, not N
• Every payload still reaches listeners — existing Watch* streams behave identically,
  with no changes needed on the listener side
• Batches are chunked to stay well under PostgreSQL's 65,535 bind-parameter limit
• A failed batch is retried, not dropped — batching must not turn one lost notification into N
• A test covers the batching path (N queued notifications → 1 round trip, all delivered)
• Ideally: a before/after number for UpdateActionStatus p95 under load

Getting started

• Reproduce: run a wide fan-out against a dev cluster and watch
  rpc_server_duration_milliseconds for UpdateActionStatus. The
  flyteorg/benchmark suite does this —
  uv run scripts/v2/swarm.py --k 25 --n 2000 generates ~100 action-updates/second.
• See the ceiling yourself: against any Postgres, compare
  DO $$ BEGIN FOR i IN 1..1000 LOOP PERFORM pg_notify('probe','x'); END LOOP; END $$;
  (one round trip) with 1,000 separate SELECT pg_notify('probe','x'); statements.
• Relevant code: start at runNotifyLoop — the pump is about 70 lines.
• How to test: go test ./runs/repository/impl/... — needs a local Postgres on port 15432
  (user/password: postgres, db flyte_runs_test; see testDbConfig in action_test.go).
• Setup: CONTRIBUTING.md

Related — three issues on the same bottleneck:

• perf: batch pg_notify round trips in the runs-service NOTIFY pump #7756 batch the pump's round trips (this one)
• perf: don't block action writes on the NOTIFY pump #7757 don't block action writes on the pump
• feat: add metrics for the runs-service NOTIFY pump (pending depth, notify latency, coalescing) #7758 instrument the pump