[ARO-25354] Monitor fixes for races

[hawkowl]

Fixes ARO-25354

Reduces test races in the Monitor and reduces flakes.

[openshift-ci]

Skipping CI for Draft Pull Request.
If you want CI signal for your change, please convert it to an actual PR.
You can still manually trigger a test run with /test all

[github-actions]

Please rebase pull request.

[github-actions]

Please rebase pull request.

[tuxerrante]

Could we guard this close for nil? For non-owned buckets, fixDoc can leave v.stop == nil (no worker started), but deleteDoc can still be reached for that cached doc, and close(nil) would panic. The previous implementation only closed when non-nil. Related context:

ARO-RP/pkg/monitor/cache.go

Lines 150 to 160 in 0c3abcb

	func (mon *clusterChangeFeedResponder) fixDoc(v *cacheDoc) {
	_, ours := mon.buckets[v.doc.Bucket]
	if !ours && v.stop != nil {
	close(v.stop)
	v.stop = nil
	} else if ours && v.stop == nil {
	ch := make(chan struct{})
	v.stop = ch
	go mon.newWorker(ch, v.doc.ID)
	}

[Copilot]

Pull request overview

This PR addresses ARO-25354 by reducing race conditions and flakiness in Monitor-related changefeed processing, primarily by tracking “processed” vs “data updated” timestamps and by refactoring the monitor’s cluster cache to use a concurrent map.

Changes:

• Extend the generic changefeed consumer contract to report whether any documents were retrieved in the current polling iteration.
• Add “last data update” timestamps to changefeed-backed caches and update tests to wait on those signals instead of fake iterator consumption.
• Refactor Monitor’s cluster cache into a dedicated changefeed responder backed by xsync.Map.

Reviewed changes

Copilot reviewed 13 out of 13 changed files in this pull request and generated 9 comments.

Show a summary per file

File	Description
pkg/util/changefeed/subscriptioncache.go	Adds GetLastDataUpdate and tracks processed vs data-update timestamps.
pkg/util/changefeed/subscriptioncache_test.go	Replaces iterator-consumption waits with timestamp-based waits.
pkg/util/changefeed/changefeed.go	Changes consumer callback to OnAllPendingProcessed(bool) and plumbs “gotAny” through.
pkg/util/changefeed/changefeed_test.go	Updates fake responder to new callback signature.
pkg/monitor/worker.go	Introduces clusterChangeFeedResponder cache/responder and updates worker to read from it.
pkg/monitor/worker_test.go	Updates changefeed wait conditions to timestamp-based signals.
pkg/monitor/monitor.go	Wires the new cluster responder into Monitor and readiness checks.
pkg/monitor/monitor_test.go	Updates bucket assertions to use the new responder state.
pkg/monitor/cache.go	Moves cache doc lifecycle (upsert/delete/fix) onto the new responder with xsync.Map.
pkg/monitor/cache_test.go	Updates cache tests to use the new responder and map.
pkg/mimo/scheduler/clustercache.go	Aligns MIMO cluster cache changefeed timestamp tracking with the new “processed vs updated” split.
pkg/database/cosmosdb/subscriptions_ext.go	Removes fake client AllIteratorsConsumed helper (tests no longer rely on it).
pkg/database/cosmosdb/openshiftcluster_ext.go	Removes fake client AllIteratorsConsumed helper (tests no longer rely on it).

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated 1 comment.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[hawkowl]

/azp run ci

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).