fix: reader survives raising apply-observer callbacks and transient catch-up unassignment

docs/API.md

@@ -75,7 +75,7 @@ The reader and writer fix the key byte-layer at UTF-8 and expose only
 |---|---|
 | `ensure_topic(bootstrap_servers, topic, *, num_partitions=1, replication_factor=1, topic_configs=None, on_policy_mismatch="warn") -> EnsureTopicResult` | Idempotent explicit create, plus the policy check/reconcile below. Defaults are dev-grade — production registries want RF≥3, `min.insync.replicas=2`, `acks=all`. |
 | `DEFAULT_TOPIC_CONFIGS` | `{"cleanup.policy": "compact"}` (read-only mapping). |
-| `ViewStats` | Frozen counters: `records_applied`, `tombstones_applied`, `keyless_records`, `key_decode_errors`, `value_decode_errors`, `catch_up_seconds`, `replayed_at_catch_up`. |
+| `ViewStats` | Frozen counters: `records_applied`, `tombstones_applied`, `keyless_records`, `key_decode_errors`, `value_decode_errors`, `callback_errors`, `catch_up_seconds`, `replayed_at_catch_up`. |
 | `SupportsJsonModel` | Protocol the `.json()` presets require (`model_dump_json` / `model_validate_json`). |
 | `TableStatus` | The `status` literal type. |
 

docs/adr/0003-catch-up-gate-tolerates-transient-unassignment.md

@@ -0,0 +1,68 @@
+# Catch-up gate tolerates a transient partition unassignment instead of failing the reader
+
+**Status:** accepted
+
+Under `group_id=None`, partition assignment is not fixed: `NoGroupCoordinator`
+re-derives it on every metadata update. A metadata blip — most commonly a
+`LeaderNotAvailableError` during a leader election or broker restart — momentarily
+drops the subscribed topic from cluster metadata, so the assignment is briefly
+replaced with an empty one and `consumer.position(tp)` raises `IllegalStateError`
+until the next metadata refresh restores it. The reader loop's catch-up gate read
+(`positions = [await consumer.position(tp) for tp in tps]`) was unguarded, so a
+*self-healing* blip that landed while the table was still catching up killed the
+reader permanently (`status == "failed"`, frozen for the object's lifetime). The
+`barrier()` sweep already tolerated this exact exception; the catch-up gate did
+not. We make the catch-up gate tolerate it too: catch `IllegalStateError`, defer
+the gate for that iteration (`positions = None`), and latch only on a complete
+successful read. aiokafka retries the blip away and the reassigned partitions
+replay from earliest, so the fixed start-time end-offset gate re-evaluates and
+latches on a later pass.
+
+## Considered options
+
+- **Leave the catch-up read unguarded (fail-fast)** — rejected: it turns a
+  transient, self-healing condition into a permanent, unrecoverable death, and it
+  is inconsistent with the `barrier()` path, which already defers on the very same
+  `IllegalStateError` (with a comment calling it "a blip aiokafka self-heals
+  from"). A sibling read guarding an exception that its twin lets kill the reader
+  is an oversight, not a decision.
+- **Distinguish transient from permanent unassignment (bounded defer, then fail)**
+  — rejected: aiokafka gives no signal at the point of the raise to tell a
+  momentary blip from a permanent unassignment (e.g. a deleted topic). The
+  existing `catchup_timeout` already bounds an indefinite defer and converts it
+  into a loud `degraded` state, which is a better terminal than `failed` for a
+  condition that may still recover.
+- **Broaden the catch to `KafkaError`/`Exception`** — rejected as dangerous:
+  `IllegalStateError` is the *only* self-healing signal here. Anything else
+  (including a genuine broker fault surfaced through `position()`'s
+  `check_errors()`) must still kill the reader loudly. Broadening would convert a
+  real fault into an infinite, silent `degraded` spin.
+
+## Consequences
+
+- **Transient blip mid-catch-up now recovers** instead of failing: the gate
+  defers, the assignment self-heals, and the reader latches on a later iteration.
+- **Permanent unassignment now degrades, not fails.** A permanently-raising
+  `position()` (e.g. the topic is deleted mid-catch-up) defers forever, reaches
+  `catchup_timeout`, and surfaces as `status == "degraded"` (alive, recoverable,
+  loudly logged once at `start()`), never `failed`. This qualifies the library's
+  "non-retriable error latches `failed`" posture: a non-retriable *fetch* error
+  (e.g. `TopicAuthorizationFailedError`) still fails, but a permanent *catch-up
+  unassignment* degrades — the same defer-not-fail trade-off `barrier()` makes.
+- **No false-positive catch-up.** Deferring sets `positions = None`, and the latch
+  is gated on `positions is not None and all(...)`, so the gate can only latch on
+  a complete successful read — never on partial or blipped data. The gate stays
+  all-or-nothing across partitions (one partition blipping defers the whole gate),
+  deliberately distinct from `barrier()`'s per-`tp` skip.
+- **A genuine fault still dies loudly.** A non-`IllegalStateError` from the
+  catch-up `position()` propagates out of the reader task and latches `failed`.
+- **Observability trade-off:** the `degraded` transition is logged once at ERROR
+  by `start()` and is queryable via `status`/`stats`; there is no periodic re-log
+  while permanently wedged. Accepted as-is — it matches the pre-existing
+  serve-degraded-don't-crash-loop posture and keeps the wedged state honestly
+  reflected in `status`.
+- **Verified** by broker-free unit tests driving the reader loop with a fake
+  consumer whose `position()` raises `IllegalStateError` transiently (recovers to
+  `caught_up`), permanently (reaches `degraded`, never `failed`), and raises a
+  non-`IllegalStateError` (still `failed`); plus a multi-partition partial-blip
+  test pinning the all-or-nothing gate.

ktables/grouped_table.py

@@ -180,6 +180,10 @@ def __repr__(self) -> str:
         return f"<GroupedKafkaTable topic={self._table.topic!r} status={self._table.status} groups={len(self._index)}>"
 
     # -- index maintenance (the on_set/on_delete handlers) --------------------
+    # These are compute-then-commit: the only fallible step (codec decode) is
+    # caught in _decode_or_count, then the index mutation is pure dict work that
+    # cannot raise. So they never trip the base table's callback_errors backstop
+    # (KafkaTable._notify_observer) — that backstop exists for EXTERNAL observers.
 
     def _decode_or_count(self, key: str) -> tuple[str, str] | None:
         """Decode the composite key; count + skip foreign keys (``decode`` returns ``None``,

ktables/kafka_table.py

@@ -451,13 +451,20 @@ def _utf8_encode(s: str) -> bytes:
 
 @dataclass(frozen=True, slots=True)
 class ViewStats:
-    """An immutable point-in-time snapshot of the reader's counters."""
+    """An immutable point-in-time snapshot of the reader's counters.
+
+    ``callback_errors`` counts times an ``on_set``/``on_delete`` apply-observer
+    raised: the base table still applied the record, but the observer skipped it
+    (a raise is logged and counted, never fatal — see
+    :meth:`KafkaTable._notify_observer`).
+    """
 
     records_applied: int = 0
     tombstones_applied: int = 0
     keyless_records: int = 0
     key_decode_errors: int = 0
     value_decode_errors: int = 0
+    callback_errors: int = 0
     catch_up_seconds: float | None = None
     replayed_at_catch_up: int = 0
 
@@ -473,6 +480,7 @@ def __init__(self) -> None:
         self.keyless_records = 0
         self.key_decode_errors = 0
         self.value_decode_errors = 0
+        self.callback_errors = 0
         self.catch_up_seconds: float | None = None
         self.replayed_at_catch_up = 0
 
@@ -534,8 +542,10 @@ def __init__(
         self._key_decoder = key_decoder
         # Optional apply observers for derived views (e.g. GroupedKafkaTable):
         # on_set(key, value) fires on an applied value record, on_delete(key) on
-        # a tombstone. Synchronous, fired inside _apply after the dict mutation;
-        # must not raise (a raise kills the reader). See grouped_table.py.
+        # a tombstone. Synchronous, fired inside _apply after the dict mutation.
+        # A raise is logged and counted (stats.callback_errors), never fatal — the
+        # base record still applies but the observer misses that update, so keep
+        # observers compute-then-commit to stay consistent. See grouped_table.py.
         self._on_set = on_set
         self._on_delete = on_delete
         self._catchup_timeout = catchup_timeout
@@ -908,7 +918,7 @@ def _apply(self, record: ConsumerRecord) -> None:
             self._data.pop(key, None)
             self._live.tombstones_applied += 1
             if self._on_delete is not None:
-                self._on_delete(key)
+                self._notify_observer(self._on_delete, key, where=where)
             return
         try:
             value = self._value_decoder(record.value)
@@ -922,7 +932,30 @@ def _apply(self, record: ConsumerRecord) -> None:
         self._data[key] = value
         self._live.records_applied += 1
         if self._on_set is not None:
-            self._on_set(key, value)
+            self._notify_observer(self._on_set, key, value, where=where)
+
+    def _notify_observer(self, callback: Callable[..., None], *args: object, where: str) -> None:
+        """Fire an apply-observer hook; a raise is logged + counted, never fatal.
+
+        The base dict was already mutated by the caller, so the record is applied
+        regardless of the observer's fate. Catches ``Exception`` only — a
+        ``BaseException`` (cancellation/interrupt) still propagates, matching the
+        decoder guards above. Reader liveness is the table's contract here; the
+        observer's own integrity is its responsibility (compute-then-commit, as
+        derived views like ``GroupedKafkaTable`` do).
+        """
+        try:
+            callback(*args)
+        except Exception:
+            self._live.callback_errors += 1
+            # args[0] is the key on both paths (on_set(key, value) / on_delete(key));
+            # naming it tells operators which key the observer missed (LWW self-heals
+            # it on that key's next write) — mirrors the value-decode log's key context.
+            logger.exception(
+                "apply-observer callback raised; record applied, observer skipped it (key=%s, %s)",
+                args[0],
+                where,
+            )
 
     async def _run(
         self,
@@ -931,8 +964,11 @@ async def _run(
         end_offsets: dict[TopicPartition, int],
         started: float,
     ) -> None:
-        # Escaping exceptions are captured by _on_reader_done (status
-        # "failed"); transient outages don't raise — getmany returns empty.
+        # Escaping exceptions are captured by _on_reader_done (status "failed");
+        # transient outages don't raise — getmany returns empty. The two in-loop
+        # position() reads (catch-up gate below, barrier sweep further down) both
+        # tolerate a transient IllegalStateError from a metadata-blip assignment
+        # shrink, so a self-healing blip never latches the reader as failed.
         while True:
             batches = await consumer.getmany(timeout_ms=self._poll_timeout_ms)
             for records in batches.values():
@@ -948,8 +984,21 @@ async def _run(
                     if new:
                         end_offsets.update(await consumer.end_offsets(new))
                         tps = sorted(current, key=lambda tp: tp.partition)
-                positions = [await consumer.position(tp) for tp in tps]
-                if all(pos >= end_offsets[tp] for pos, tp in zip(positions, tps, strict=True)):
+                # A metadata blip can momentarily empty the assignment under
+                # group_id=None, making position() raise IllegalStateError. The
+                # barrier sweep below guards the same blip; guard it here too so a
+                # self-healing blip mid-catch-up defers the gate rather than killing
+                # the reader for good. Defer (positions=None) => never latch on a
+                # blip: the gate can only latch on a complete successful read.
+                # (A NON-IllegalStateError still propagates — a real fault must die,
+                # not spin.) A permanent unassignment (e.g. topic deleted) defers
+                # forever and reaches catchup_timeout => degraded, not failed —
+                # same trade-off barrier() already makes.
+                try:
+                    positions: list[int] | None = [await consumer.position(tp) for tp in tps]
+                except IllegalStateError:
+                    positions = None
+                if positions is not None and all(pos >= end_offsets[tp] for pos, tp in zip(positions, tps, strict=True)):
                     self._live.catch_up_seconds = time.perf_counter() - started
                     self._live.replayed_at_catch_up = self._live.records_applied + self._live.tombstones_applied
                     self._caught_up.set()

tests/test_grouped_table.py

@@ -487,6 +487,9 @@ async def test_reads_reflect_seeded_members_after_barrier(self, topic, grouped_t
             assert await table.barrier()
             assert await table.wait_until_caught_up()
             assert table.status == "caught_up"
+            # Grouped's own compute-then-commit hooks never raise, so the base
+            # table's callback_errors backstop stays untouched on well-formed input.
+            assert table.stats.callback_errors == 0
             assert table.get_member("billing", "a") == Endpoint(url="http://a")
             assert table.get_member("billing", "z") is None
             assert table.has_member("billing", "b") is True