fix: full PyPy support — deterministic stream cleanup + assignor livelock fix, all PyPy skips removed

[wbarnha]

Description

Makes faust run and test green on PyPy, and removes every skipif(PyPy) marker
in the test suite (21 decorators covering 30 tests, added wholesale against
PyPy 3.9 in #530/#621 and never re-validated on a modern PyPy). Two independent
root causes, both proven on a real PyPy interpreter.

---

Root cause 1 — GC-timing-dependent stream cleanup (faust/streams.py)

Cleanup for faust's stream async generators lives in finally blocks that only
run when the generator is finalized. When a caller abandons one (break
inside async for), CPython's refcounting finalizes immediately; PyPy defers
finalization to a GC cycle that may never come (PyPy CPython-differences,
PEP 525, PEP 533).
Affected duties: acking consumed events, restoring enable_acks, detaching
the buffering processor, and channel detach.

Fix (minimal): Stream tracks the async generators it hands out in a
WeakSet and closes leftovers in on_stop():

1. the take() family — take, take_events, take_with_timestamp, noack_take.
2. the main iterator from __aiter__ (_py_aiter/_c_aiter) — its inner
   finally acks the last-yielded event. (Service.stop() sets _stopped
   before awaiting on_stop(), so the iterator finally's re-entrant
   self.stop() is a guarded no-op.)

The WeakSet keeps CPython's prompt refcount finalization — and therefore all
existing CPython behavior — fully intact.

---

Root cause 2 — CopartitionedAssignor standby livelock (faust/assignor/)

_assign_round_robin frees a slot on an exhausted client by evicting an
arbitrary partition (set.pop) and re-queueing it. When that arbitrary
victim is the partition's only possible home, two partitions ping-pong between
the work queue and the same client forever. Whether it happens depends on set
iteration order, so identical input completes in ~0.3ms on CPython but livelocks
indefinitely on PyPy — the real reason the copartitioned-assignor property tests
were skipped there (not, as first assumed, mere JIT-warm-up slowness).

Minimal reproducer, deterministic even at PYTHONHASHSEED=0:

input	CPython	PyPy (before)	PyPy (after)
4 clients / 16 partitions / 1 replica	0.0003s	>25s (livelock)	0.0009s
64 / 256 / 32	0.1s	>20s	0.09s

Fix: evict a victim that some other not-yet-exhausted client can
immediately take, in deterministic (sorted) order, so the re-queued partition
makes forward progress instead of bouncing straight back. Terminates on every
interpreter and keeps the assignment valid. Any replicas >= 1 case exercises
the standby path.

---

Test-suite changes

• All 21 skipif(PyPy) decorators removed (functional/unit streams, agents,
  meticulous assignor).
• wait_for_stream_ack rewritten to close the take()/iterator generator
  explicitly (await agen.aclose()) instead of leaning on asyncio.sleep(0)
  yields — runs the acking finally deterministically on every interpreter.
• tests/conftest.py: event_loop fixture override — on PyPy, run
  gc.collect() + loop.shutdown_asyncgens() before closing each test's loop,
  so deferred finalizers execute inside the owning test instead of landing on a
  dead loop and leaking tasks/warnings into later tests.
• Regression tests: test_take_cleanup_on_stream_stop and
  test_aiter_cleanup_on_stream_stop hold a strong reference to the
  generator (blocking finalization on every interpreter, mimicking PyPy) and
  assert the stop path performs the cleanup; test_standby_assignment_terminates
  guards the assignor livelock. All verified to fail without their fix.
• The copartitioned-assignor property tests keep their original
  deadline=4000/full example budget — no PyPy-specific relaxation needed once
  the livelock is fixed (they pass in ~4s on PyPy).
• CI: PyPy leg timeout-minutes 10 → 15 (now runs the formerly-skipped tests;
  PyPy is slower and the old cap already clipped the run at ~96%).

Verification (real PyPy 3.9, NO_CYTHON=1)

• tests/functional/test_streams.py — 54 passed.
• tests/unit — 1674 passed with the strict error::ResourceWarning filter,
  zero regressions from the event_loop override.
• tests/meticulous/assignor — 4 passed in ~4s, stable across four
  PYTHONHASHSEED values.
• CPython: same suites green with Cython enabled and NO_CYTHON=1.
• flake8/black/isort clean.

🤖 Generated with Claude Code

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 94.18%. Comparing base (0a92897) to head (4ab51a5).

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #716      +/-   ##
==========================================
+ Coverage   94.15%   94.18%   +0.02%     
==========================================
  Files         104      104              
  Lines       11136    11154      +18     
  Branches     1201     1202       +1     
==========================================
+ Hits        10485    10505      +20     
+ Misses        550      548       -2     
  Partials      101      101              

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.