Skip to content

feat: implement shared watch stream - #176

Draft
smira wants to merge 2 commits into
cosi-project:mainfrom
smira:feat/shared-watch-stream
Draft

feat: implement shared watch stream#176
smira wants to merge 2 commits into
cosi-project:mainfrom
smira:feat/shared-watch-stream

Conversation

@smira

@smira smira commented Aug 3, 2026

Copy link
Copy Markdown
Member

This ensures consistency across concurrent Watch* calls vs. the watch client, mostly important for COSI runtime.

This ensures consistency across concurrent Watch* calls vs. the watch
client, mostly important for COSI runtime.

Signed-off-by: Andrey Smirnov <andrey.smirnov@siderolabs.com>
Copilot AI review requested due to automatic review settings August 3, 2026 15:26
@github-project-automation github-project-automation Bot moved this to To Do in Planning Aug 3, 2026
@talos-bot talos-bot moved this from To Do to In Review in Planning Aug 3, 2026

Copilot AI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

This PR adds an optional “shared watch” mode to the etcd-backed State, where all Watch* calls (except bookmark-resume watches) are served by a single etcd watcher plus an internal dispatcher/queueing layer to provide consistent per-destination-channel delivery ordering—primarily to satisfy COSI controller runtime expectations.

Changes:

  • Introduces a shared watcher multiplexer (watchMux) and per-destination output queues (outQueue) to serialize delivery without blocking the dispatcher.
  • Adds WithSharedWatch() state option and wires it into State.Watch / State.WatchKind* to switch between shared vs dedicated watcher paths.
  • Expands test coverage to run existing suites in both modes and adds dedicated ordering/restart tests for shared watch.

Reviewed changes

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

Show a summary per file
File Description
pkg/state/impl/etcd/watchqueue.go New per-destination delivery queue used by the shared watch dispatcher.
pkg/state/impl/etcd/watchmux.go New shared watcher multiplexer, subscription index, bootstrap splicing, and dispatch logic.
pkg/state/impl/etcd/options.go Adds sharedWatch option and WithSharedWatch() API with detailed behavior docs.
pkg/state/impl/etcd/etcd.go Adds State.mux and routes eligible Watch* calls through the shared watcher path.
pkg/state/impl/etcd/key.go Adds helper to derive kind-prefix from a full etcd key for dispatch grouping.
pkg/state/impl/etcd/watch_test.go Runs key watch regression tests in both dedicated and shared watch modes.
pkg/state/impl/etcd/watch_ordering_test.go New tests verifying cross-kind ordering, bootstrap consistency, restart, and single etcd watcher behavior under shared watch.
pkg/state/impl/etcd/state_test.go Runs state conformance suite with WithSharedWatch() enabled.
pkg/state/impl/etcd/controller_runtime_test.go Runs controller runtime conformance suite with WithSharedWatch() enabled.

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

Comment on lines +86 to +95
case <-q.done:
// deliver whatever was enqueued just before the shutdown, notably the Errored events
// reporting why the shared watcher went away
for _, item := range q.take() {
if !item.sub.send(item.events) {
break
}
}

return
Comment on lines +418 to +420
for _, sub := range subs {
sub.deliver(converted)
}
Pull in new etcd, update Go deps, rekres.

Signed-off-by: Andrey Smirnov <andrey.smirnov@siderolabs.com>
@smira
smira marked this pull request as draft August 4, 2026 12:29
@smira
smira marked this pull request as draft August 4, 2026 12:29
@smira smira moved this from In Review to To Do in Planning Aug 4, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants