Add three Cython accelerators: offset commit, record scheduler, sensor fan-out

[wbarnha]

Description

Faust already ships optional Cython implementations of the window types (faust/_cython/windows.pyx), the stream iterator (faust/_cython/streams.pyx) and the topic conductor (faust/transport/_cython/conductor.pyx). This adds three more, found by profiling the per-message and per-commit work.

Each follows the pattern already in the tree: the pure-Python implementation stays and is used whenever the extension could not be built or NO_CYTHON is set. Nothing here is required for Faust to run.

What was added

faust/utils/_cython/functional.pyx — first_consecutive_run()

Consumer._new_offset only needs the first run from consecutive_numbers(), but the groupby() implementation builds a tuple, calls a Python key function and creates a group generator for every acked offset. It runs once per assigned partition on every commit, and because it blocks the event loop the cost shows up as latency rather than throughput.

faust/transport/_cython/scheduler.pyx — records_iterator()

A C-level cursor replacing the round-robin generator returned by DefaultSchedulingStrategy.records_iterator, which also takes over TopicBuffer's inner per-partition generator. Two generator frames were being resumed for every record fetched from the broker.

faust/sensors/_cython/base.pyx — SensorDelegateBase

A cdef base class carrying the four sensor hooks that fire on every message. SensorDelegate subclasses it and keeps its ~20 occasional hooks in plain Python, so only the hot quarter of the class moved.

Benchmarks

extra/tools/benchmark_cython.py (new) times each accelerator against its pure-Python counterpart in the same interpreter. On CPython 3.11:

benchmark	python	cython	speedup
first_consecutive_run / 100 offsets	3.10 µs	0.78 µs	3.95x
first_consecutive_run / 10k offsets	269.6 µs	52.0 µs	5.18x
first_consecutive_run / 100k offsets	2714.1 µs	551.6 µs	4.92x
records_iterator / 1 topic × 1 part	500.6 ns/rec	224.8 ns/rec	2.23x
records_iterator / 4 topics × 8 parts	325.7 ns/rec	123.9 ns/rec	2.63x
records_iterator / 8 topics × 16 parts	323.5 ns/rec	116.6 ns/rec	2.77x
SensorDelegate.on_stream_event_in / 1 sensor	373.7 ns	181.9 ns	2.05x
SensorDelegate all 4 hooks / 1 sensor	745.7 ns	486.2 ns	1.53x

Three honest caveats:

• The records_iterator figures were previously 5.7–7.7x here. That version skipped rebuilding its snapshot when the mapping length was unchanged, which turned out to drop records when a buffer was replaced under an existing key (see below). Correctness cost most of the win; these numbers are the fixed implementation. An unconditional rebuild was also measured, at 1.6–2.1x, so the current identity check is worth keeping over simply rebuilding.
• For first_consecutive_run, part of the win is just dropping groupby — the new pure-Python helper is already ~2.6x faster than the old expression, and Cython adds ~4-5x on top of that. The table compares the two new implementations, not old-vs-new.
• The sensor numbers use a no-op sensor, so they measure delegation overhead only. With a real Monitor attached, the sensor bodies dominate and the relative win is smaller.

Behaviour

Preserved rather than approximated. Specifically:

• Both scheduler implementations re-read the topic index and each topic's buffer map at every pass and pop drained entries from them, so adding, removing or replacing a topic or partition mid-iteration gives the same result either way, and TopicBuffer._buffers is drained identically.
• A TopicBuffer subclass is still driven through next(), so an overridden __iter__/__next__ is honoured. map_from_records and records_iterator remain overridable, so a custom ConsumerScheduler is unaffected.
• first_consecutive_run stops consuming as soon as the run ends, so a shared iterator is left exactly where the pure-Python version leaves it. It re-reads the list length while scanning, because a custom __sub__ can resize the list underneath it.
• The sensor set is read live rather than snapshotted, so mutating _sensors directly behaves the same in both.

Not covered deliberately: swapping in a topic under a different name mid-iteration. The pure-Python generator iterates the live dict, so CPython raises dictionary keys changed during iteration — undefined behaviour in Python itself rather than a guarantee to pin.

Testing

• New tests run every code path against both implementations, including a randomised differential test over 200 scheduler topologies and regression tests for each behaviour above.
• Full suite passes in all three configurations: extensions built (2311 passed), NO_CYTHON=1 (2305 passed), and extensions absent entirely.
• flake8 / black / isort clean; sdist ships all six .pyx files and rebuilds from a clean tree.

Evaluated and rejected

Recorded here so the ground does not get re-covered:

• Message/ConsumerMessage as cdef classes — cdef public compiles to getset descriptors, which CPython 3.11+ cannot specialize the way it specializes the __slots__ member descriptors Message already uses. Most readers are pure Python, so attribute reads would likely regress. Only worth it bundled with a tuples.pxd so the existing .pyx modules can cimport it.
• Event — EventT is Generic[T], AsyncContextManager, so cdef class Event(EventT) does not compile. The EventT.register(Event) workaround routes the per-message isinstance checks through ABCMeta.__instancecheck__, costing more than the constructor saves.
• Registry (serializers) — real dispatch waste, but most of it is recoverable in pure Python; Cython's marginal contribution is small.
• Schema, Codec, Table/Collection, Monitor, Consumer — all blocked by Service/ABC bases, documented subclassing, or both.
• Record codegen and faust/models/typing.py — generated via exec() at class-definition time; Cython cannot apply.
• faust/utils/iso8601.py — already delegates to ciso8601 when installed.

FieldDescriptor (faust/models/fields.py) is the one genuinely promising candidate left — ~30 descriptor calls per typed-model message — but it needs a hybrid cdef base plus converting two cached_property attributes, which is a larger change than this PR. Left as follow-up.

Note on the assignor commit

2f04895 is a cherry-pick of the existing fix on claude/fix-pypy-hypothesis-deadline (which has no open PR). Without it the pypy3.11 leg fails on a hypothesis deadline flake in test_copartitioned_assignor.py that is unrelated to this diff — see the comment below for the evidence. Happy to drop it if that branch lands separately.

🤖 Generated with Claude Code

https://claude.ai/code/session_019oX4oGCQGgvPJHabjwfaBk

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 96.06%. Comparing base (803c7a4) to head (340793f).

Additional details and impacted files

@@           Coverage Diff           @@
##           master     #751   +/-   ##
=======================================
  Coverage   96.06%   96.06%           
=======================================
  Files         103      103           
  Lines       11072    11094   +22     
  Branches     1191     1193    +2     
=======================================
+ Hits        10636    10658   +22     
  Misses        345      345           
  Partials       91       91           

☔ 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.

[wbarnha]

The Python pypy3.11/Cython: false leg was failing on a flake unrelated to this PR, so I picked up the existing fix for it — flagging that here since it explains why an assignor-test commit is in a Cython PR.

The failure was test_remove_clients in tests/meticulous/assignor/test_copartitioned_assignor.py:

hypothesis.errors.FlakyFailure: ... produces unreliable results:
Failed on the first call but did not on a subsequent one
Unreliable test timings! On an initial run, this test took 5061.03ms, which
exceeded the deadline of 4000.00ms, but on a subsequent run it took 3483.45ms,
which did not.

Evidence it isn't caused by this branch:

• It reproduced identically on Evaluate a feature-flagged Rust accelerator build #749 — same test, same generated example (partitions=243, replicas=62, num_clients=686, num_removal_clients=1), 5154ms vs the same 4000ms deadline. That branch's unique content is documentation and a Rust crate, neither of which can affect the assignor.
• Nothing in either diff touches faust/assignor/ or tests/meticulous/.
• The test passes on retry — it is PyPy JIT warmup pushing one example over a fixed deadline, not an assignment-correctness failure.

Hypothesis shrinks to the same slow example every run, so it was not going to clear on its own. 2f04895 is a cherry-pick of the existing fix on claude/fix-pypy-hypothesis-deadline (TEST_DEADLINE = None), which has no open PR — so it carries the original commit and authorship rather than a re-write. Happy to drop it from here if you would rather land that branch separately.

Everything else is green, and codecov/patch / codecov/project reported all modified lines covered once the full set of coverage uploads landed — the earlier red was a partial-upload artifact while the matrix was still running.

---

Generated by Claude Code

[wbarnha]

Confirmed and fixed in 340793f. Thank you — this was a real record-loss bug, and the diagnosis was exactly right.

Reproduced before touching anything, against the pure-Python iterator as the reference:

python: [1, 10, 11]
cython: [1, 2]        <- 10 and 11 never delivered

The length check could not distinguish "unchanged" from "swapped", so the shortcut was taken and the stale cursor kept draining the old buffer. The cursor.source is not buffer check that exists to catch precisely this lived inside the rebuild that the shortcut skipped.

The same flaw was present one level down, which the report did not mention: in _TopicCursor._start_pass, an iterator swapped in under an existing TP left the partition cursor stale the same way. Same reproduction, same silent loss. Both are fixed.

On the suggested remedies — I took the second one, but not by rebuilding unconditionally. Both shortcuts are now _snapshot_is_current(), which compares the cached snapshot against the live mapping entry by entry by identity. It walks the mapping but allocates nothing, and that distinction matters:

variant	1t × 1p	8t × 16p	correct
length-only shortcut (before)	7.71x	5.84x	❌
unconditional rebuild	1.62x	2.05x	✅
identity check (shipped)	2.23x	2.77x	✅

So the fix costs most of the speedup — the scheduler was 5.7–7.7x over pure Python and is now 2.2–2.8x. A pass has to consult the live mapping to be correct, and a single-partition topic reaches a pass boundary on every record. I have corrected the benchmark table in the PR description rather than leave it overstating the result.

Both of your regression scenarios are now tests, parametrised over both implementations, plus the partition-level analogue. The randomised differential test is at 4000/4000.

One scenario I deliberately did not pin: replacing a topic under a different name. The pure-Python generator iterates the live dict, so CPython raises dictionary keys changed during iteration there — that is undefined behaviour in Python itself, not a guarantee either implementation should be held to. Noted in the test file.

---

Generated by Claude Code