watcherGroup.choose has two branches. When the unsynced group holds fewer than
maxWatchersPerSync watchers it calls chooseAll on the group itself. When it holds at least
that many, it copies a bounded subset into a temporary group and calls chooseAll on the copy:
func (wg *watcherGroup) choose(maxWatchers int, curRev, compactRev int64) (*watcherGroup, int64) {
if len(wg.watchers) < maxWatchers {
return wg, wg.chooseAll(curRev, compactRev)
}
ret := newWatcherGroup()
for w := range wg.watchers {
if maxWatchers <= 0 {
break
}
maxWatchers--
ret.add(w)
}
return &ret, ret.chooseAll(curRev, compactRev)
}
chooseAll is the only place that retires a watcher whose minRev is below compactRev: it
sends a CompactRevision response, sets w.compacted = true, and calls wg.delete(w). In the
capped branch that wg is the temporary group, so the watcher is removed from the copy and left
in s.unsynced.
syncWatchers then iterates the returned group, which no longer contains the watcher, so it
never reaches s.unsynced.delete(w) either. Effects:
- the watcher stays in the unsynced group until the client cancels it
- every following resync selects it again and sends another
CompactRevision response on the
same stream, once per watchResyncPeriod (100ms)
s.unsynced.size() stays inflated, and that value is both the return of syncWatchers and the
input to slowWatcherGauge, so the resync loop's own backpressure signal is wrong
Reachable whenever at least maxWatchersPerSync (default 512) watchers are unsynced at the same
time and at least one of them starts below the compaction revision. That is a normal state on a
busy cluster after a compaction, and also right after a leader change or snapshot restore, since
watchableStore.Restore moves every synced watcher into the unsynced group at once.
Impact is availability and noise rather than data loss: no events are lost or duplicated, but a
watch that should have been cleanly retired is not, and the client receives repeated compaction
responses.
Existing coverage does not reach it. TestWatchNoEventLossOnCompact sets
maxWatchersPerSync = 4 with 3 watchers, so it only ever exercises the uncapped branch. No test
in the package drives choose with len(watchers) >= maxWatchers and compacted watchers present.
watcherGroup.choosehas two branches. When the unsynced group holds fewer thanmaxWatchersPerSyncwatchers it callschooseAllon the group itself. When it holds at leastthat many, it copies a bounded subset into a temporary group and calls
chooseAllon the copy:chooseAllis the only place that retires a watcher whoseminRevis belowcompactRev: itsends a
CompactRevisionresponse, setsw.compacted = true, and callswg.delete(w). In thecapped branch that
wgis the temporary group, so the watcher is removed from the copy and leftin
s.unsynced.syncWatchersthen iterates the returned group, which no longer contains the watcher, so itnever reaches
s.unsynced.delete(w)either. Effects:CompactRevisionresponse on thesame stream, once per
watchResyncPeriod(100ms)s.unsynced.size()stays inflated, and that value is both the return ofsyncWatchersand theinput to
slowWatcherGauge, so the resync loop's own backpressure signal is wrongReachable whenever at least
maxWatchersPerSync(default 512) watchers are unsynced at the sametime and at least one of them starts below the compaction revision. That is a normal state on a
busy cluster after a compaction, and also right after a leader change or snapshot restore, since
watchableStore.Restoremoves every synced watcher into the unsynced group at once.Impact is availability and noise rather than data loss: no events are lost or duplicated, but a
watch that should have been cleanly retired is not, and the client receives repeated compaction
responses.
Existing coverage does not reach it.
TestWatchNoEventLossOnCompactsetsmaxWatchersPerSync = 4with 3 watchers, so it only ever exercises the uncapped branch. No testin the package drives
choosewithlen(watchers) >= maxWatchersand compacted watchers present.