Speed up Celery task publishing with Redis pipelines

[shivaam]

CeleryExecutor currently publishes each task to Redis in a separate network
round trip. This makes publication time grow with both the number of scheduled
tasks and broker latency.

This change lets Celery and Kombu prepare each task message normally, then
buffers the final Redis queue writes in a non-transactional pipeline and
flushes them together. Callback workloads and non-Redis brokers keep the
existing publication path.

Redis pipelining is enabled by default and can be disabled with
[celery] redis_pipelined_publish_enabled = False. The implementation mirrors
Kombu's private Redis transport methods, so a focused contract test will fail
if a future Kombu update changes that integration point.

Local validation:

• Celery executor unit module: 74 passed, 6 skipped.
• Linux multiprocessing regression: passed 250 repeated dispatch cycles.
• PostgreSQL/Celery integration suite: 10 passed with real Redis and RabbitMQ
  brokers and Celery workers.
• Real daemon load: scheduler, API server, DAG processor, triggerer, PostgreSQL,
  Redis, and a four-process Celery worker completed 1,000/1,000 Bash tasks and
  four/four DAG runs successfully. Redis recorded exactly 1,000 queue writes
  and drained to zero.
• Changed source files pass mypy; regular prek checks pass. The
  provider-wide manual mypy hook was terminated by the local Docker resource
  limit (exit 137/143).

closes: #8854

---

Was generative AI tooling used to co-author this PR?

• Yes — Codex (GPT-5)

Generated-by: Codex (GPT-5) following the guidelines

---

Drafted-by: Codex (GPT-5) (no human review before posting)

[eladkal]

Can you clarify if this change is only for Redis or did you mention Redis because this is the broker you used for testing? Would that work with other brokers like SQS or RabbitMQ?

[shivaam]

This optimization is Redis-only because it relies on Redis pipelines. RabbitMQ and SQS continue using the existing publication path, so their behavior remains unchanged.

---

Drafted-by: Codex (GPT-5); reviewed by @shivaam before posting

[eladkal]

This optimization is Redis-only because it relies on Redis pipelines. RabbitMQ and SQS continue using the existing publication path, so their behavior remains unchanged.

Drafted-by: Codex (GPT-5); reviewed by @shivaam before posting

Lets maybe add a table to the doc that clarifies which broker supports this