🐛 eventing: the group mirror re-applies forged member terminals on replay #956

Summary

GroupMirror re-applies forged member terminals on every start, so a batch reads back failed: 1 for an agent that succeeded. This is the residual of #885 that the fix in #955 (d7d2c5f7) deliberately did not reach, named in the review there and scoped in a comment rather than widened.

Narrower than #885 — the batch stays open, because complete_group is guarded and maybe_complete is skipped on replay — but it is the same corruption of member status and counts.

Why the #885 fix does not cover it

#885 made a rejected envelope inert by marking it with ce.KEY_REJECTED. That marker is local: it is set on the dict in kafka_in after verification, and it is deliberately not on the wire (to_kafka_binary and from_kafka_binary both refuse __-prefixed names, so a forger cannot set it either).

The mirror reads the topic, not the store. So on replay there is no marker, ce.is_rejected is false, and GroupService.on_member_event does its state writes before the replay check:

store.add_group_member(groupid, corr)              # members_seen moves
final = str(event.get("final", "false")).lower() == "true"
if not final:
    store.mark_member_running(groupid, corr)
    return
first = store.mark_member_finished(groupid, corr, event.get("phase") or "result")
if replay:
    return                                          # <- only SETTLING is skipped

mark_member_finished maps terminal_phase == "error" to status failed, and wins on finished_utc IS NULL. A forged terminal lying earlier in the topic than the genuine answer therefore wins that race again on every start.

Why it is worst where the mirror matters most

Conditional in an unlucky direction:

Those are the deployments kafka_group_mirror.py was written for — its module docstring is about a restarted pod with an emptyDir /data serving 404 for every earlier group.

Why widening the mirror's verification is the wrong fix

Considered and rejected in #955:

• Verifying member events on replay needs the runner keyset with require=True, which refuses every pre-signing record — and replaying history that predates signing is the mirror's whole purpose. Every restart in an unsigned deployment would lose all group state, which is worse than the bug.
• "Verify only when a signature is present" is not a control, because a forger simply omits the signature.

Suggested fix

A replay that cannot regress a member already terminal in the store — a store-level change, not a mirror one. mark_member_finished already guards on finished_utc IS NULL; what is missing is a replay-safe variant that also refuses to move a member whose stored terminal came from a verified row, or an ordering rule that prefers the verified terminal regardless of topic position.

Note the interaction with the rejected column added in #955: the store now records which rows were refused, so the store does have the information the wire lacks. That is the most likely place to resolve this.

Where it is recorded

eventing/eventbridge/kafka_group_mirror.py:88-109 — the What this does NOT cover paragraph and its VERIFY comment, which names this fix and says which sentence to keep when it lands.

Out of scope

Not a demo blocker for KubeCon M2 (#896): the forged-event beat asserts the live path, where #885's fix holds. Worth doing before anyone demos a restart after a forgery.