bugfix: stop pinning beat dispatch to a single producer connection

[david-note]

What & why

_maybe_due_kwargs (redbeat/schedulers.py) was a cached_property returning
{'producer': self.producer}. self.producer is itself a cached_property
on celery.beat.Scheduler, wrapping a single kombu connection
(Producer(self._ensure_connected(), ...)).

Caching _maybe_due_kwargs on top of that meant a beat process resolves one
producer/connection on its first tick and reuses it for every dispatch for
the rest of the process's life — nothing ever refreshes it.

If that connection ever breaks (an idle connection reaped by the broker, a
proxy failover, a TCP reset), every subsequent publish fails. That failure is
invisible: celery.beat.Scheduler.apply_async reserves/persists
last_run_at / total_run_count / the schedule ZSET score via reserve()
before attempting the publish, and RedBeat's maybe_due catches the publish
exception and only logs it:

try:
    result = self.apply_async(entry, **kwargs)
except Exception as exc:
    logger.exception('Scheduler: Message Error: %s', exc)

RedBeat's own bookkeeping runs over a separate redis-py client that
reconnects independently, so Redis-side signals (total_run_count,
last_run_at, the ZSET score, lock TTL) keep advancing normally and look
identical whether the publish succeeded or is failing on every tick. The only
symptom is that tasks silently stop reaching workers.

Fix

Make _maybe_due_kwargs a plain @property instead of a cached_property.
This is the only change needed — it stops pinning the resolved producer, so
each tick re-reads self.producer. (self.producer itself is still a
cached_property from celery.beat.Scheduler, matching upstream Celery's
own reuse-a-producer behavior for a healthy connection; this fix only removes
RedBeat's own extra layer of caching on top of it.)

Test plan

• Added test_maybe_due_kwargs_reflects_a_fresh_producer_each_tick to
  tests/test_scheduler.py, which patches producer to return different
  mocks on successive accesses and asserts _maybe_due_kwargs['producer']
  reflects each one. Verified it fails against the pre-fix cached_property
  and passes with the fix.
• make lint and make test (90 tests) both pass.

Heads-up for reviewers

No behavior change for the common case where the connection stays healthy —
self.producer is unaffected and still cached at the celery.beat.Scheduler
level. This only removes the extra pinning RedBeat added on top, so a broken
connection can heal on the next tick instead of failing for the life of the
process.

[sibson]

Thanks for the detailed writeup — the failure mechanics you describe are real, and I verified them locally. Before merging though, I dug into whether the fix changes runtime behavior, and I don't think it does. Sharing the full analysis so we can figure out the right path from here.

What checks out

Your description of the symptom is accurate, and I confirmed it empirically (real tick() path against a memory:// broker with a sabotaged Channel.basic_publish, fakeredis for bookkeeping — no mocks on the scheduler itself):

• celery.beat.Scheduler.apply_async calls reserve() before publishing, so total_run_count, last_run_at, and the ZSET score all advance even when the publish fails.
• maybe_due swallows the exception and logs one ERROR line. The failed run is permanently lost — it's skipped, not deferred.
• RedBeat's bookkeeping goes over a separate redis-py client, so Redis-side state is indistinguishable from success. That part of the "invisible" story is exactly right.

Where the fix falls short

Scheduler.producer on the celery side is itself a cached_property, and nothing in celery or RedBeat ever invalidates it. So with this patch applied, every access to _maybe_due_kwargs re-reads self.producer… and gets the same object back out of the instance __dict__. I checked this on your branch with a live scheduler:

producer identical across accesses: True
producer cached in instance __dict__: True   (by celery, not redbeat)

The producer is exactly as pinned after the patch as before it. The new test passes because PropertyMock replaces celery's cached producer with an uncached one — it fabricates the refresh behavior the fix is meant to create, rather than observing it.

What the forced-failure experiments showed

scenario	result
transient recoverable error	kombu's publish(retry=True) retried within the same tick, revived the connection in place on the pinned producer (Connection.ensure), message delivered, nothing logged above INFO
persistent error	one ERROR Scheduler: Message Error: … per tick; bookkeeping advances anyway; run lost
broker recovers	the same pinned producer object resumes publishing on the next tick

So a pinned producer isn't inherently wedged: for errors the transport classifies as recoverable, kombu heals the connection without needing a fresh producer. A permanent wedge requires an error outside the transport's recoverable_connection_errors — and that set is transport-specific (e.g. the redis transport treats OSError/ConnectionResetError as recoverable; amqp is similar).

Where I'd love to go from here

The most valuable next step is empirical evidence that the wedge actually occurs, and a reproduction that doesn't rely on mocking:

1. Production logs. Every failed dispatch writes exactly one ERROR Scheduler: Message Error line (logger celery.beat). If you've seen this in a real deployment, the signature to look for is an unbroken run of that line, one per tick, from some moment onward — isolated occurrences point to transients that kombu healed or lost once. The exception class in the traceback would tell us whether it fell outside recoverable_connection_errors, which is the crux.
2. Repro against a real broker. A docker broker with a TCP proxy in between (e.g. toxiproxy) lets us inject the realistic failure modes without mocks: connection resets, timeouts, and especially half-open connections (drop packets rather than close — the classic case where a publish hangs or errors oddly instead of failing cleanly). rabbitmqctl close_all_connections / broker restarts cover the cleaner variants. The question each time: does the pinned producer heal on the next tick, or wedge?
3. Instrumentation so Redis stops lying. Since the defining symptom is "Redis looks healthy while publishes fail," we could record publish failures into Redis from maybe_due's except path (per-entry hash with last error, timestamp, run count). That gives any deployment a countersignal to compare against total_run_count, and would let us confirm or rule this out in the wild.
4. If a wedge is confirmed, the fix that actually reaches it is invalidating celery's caches on failure — del self.producer / del self.connection in the except path so the next tick rebuilds via _ensure_connected(). That's testable without patching producer itself.

If you hit this in production and have logs (or the broker/transport details), I'd genuinely like to dig into it with you — and I'm happy to pair on the toxiproxy repro or the instrumentation piece as a starting point. The diagnosis effort here is appreciated; I just want to make sure what we merge actually moves the needle on it.

---

Generated by Claude Code