From 29de7261bbf155460640241a86a4b32d5ef80e68 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ren=C3=A9=20Jochum?= Date: Fri, 7 Aug 2026 18:01:45 +0200 Subject: [PATCH 1/4] Revert "client: Add EventListener.SetOrdered" MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This reverts commit 0a97515f05aa649cbc9e40757df628a72e6e1c38. Signed-off-by: René Jochum --- client/events.go | 84 ++---------------------------------------- client/incus_events.go | 12 +++++- 2 files changed, 15 insertions(+), 81 deletions(-) diff --git a/client/events.go b/client/events.go index 07ed43c4b78..86d3da3b0e9 100644 --- a/client/events.go +++ b/client/events.go @@ -3,15 +3,11 @@ package incus import ( "context" "errors" - "slices" "sync" "github.com/lxc/incus/v7/shared/api" ) -// eventQueueSize is how many events may be pending handling before an ordered listener is dropped. -const eventQueueSize = 1000 - // The EventListener struct is used to interact with an Incus event stream. type EventListener struct { r *ProtocolIncus @@ -23,9 +19,6 @@ type EventListener struct { projectName string targets []*EventTarget targetsLock sync.Mutex - - // queue is only set when ordered delivery was requested. - queue chan api.Event } // The EventTarget struct is returned to the caller of AddHandler and used in RemoveHandler. @@ -34,75 +27,6 @@ type EventTarget struct { types []string } -// SetOrdered makes the listener call its handlers one event at a time and in order (must be called before AddHandler). -func (e *EventListener) SetOrdered() { - e.targetsLock.Lock() - defer e.targetsLock.Unlock() - - if e.queue != nil { - return - } - - e.queue = make(chan api.Event, eventQueueSize) - - go e.dispatch() -} - -// dispatch delivers queued events to the handlers, one event at a time. -func (e *EventListener) dispatch() { - for { - var event api.Event - - select { - case <-e.ctx.Done(): - return - case event = <-e.queue: - } - - e.targetsLock.Lock() - targets := slices.Clone(e.targets) - e.targetsLock.Unlock() - - for _, target := range targets { - if target.types != nil && !slices.Contains(target.types, event.Type) { - continue - } - - target.function(event) - } - } -} - -// send passes an event on to the handlers of this listener. -func (e *EventListener) send(event api.Event) { - e.targetsLock.Lock() - defer e.targetsLock.Unlock() - - if e.ctx.Err() != nil { - return - } - - if e.queue == nil { - for _, target := range e.targets { - if target.types != nil && !slices.Contains(target.types, event.Type) { - continue - } - - go target.function(event) - } - - return - } - - select { - case e.queue <- event: - default: - // Dropping events would leave the handler with a silently incomplete view. - e.err = errors.New("Event handlers are too far behind") - e.ctxCancel() - } -} - // AddHandler adds a function to be called whenever an event is received. func (e *EventListener) AddHandler(types []string, function func(api.Event)) (*EventTarget, error) { if function == nil { @@ -154,6 +78,10 @@ func (e *EventListener) Disconnect() { e.r.eventListenersLock.Lock() defer e.r.eventListenersLock.Unlock() + if e.ctx.Err() != nil { + return + } + // Locate and remove it from the global list for i, listener := range e.r.eventListeners[e.projectName] { if listener == e { @@ -164,10 +92,6 @@ func (e *EventListener) Disconnect() { } } - if e.ctx.Err() != nil { - return - } - // Turn off the handler e.err = nil e.ctxCancel() diff --git a/client/incus_events.go b/client/incus_events.go index f19fa599bbc..5e559d3f7a7 100644 --- a/client/incus_events.go +++ b/client/incus_events.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "net/url" + "slices" "strings" "time" @@ -177,7 +178,16 @@ func (r *ProtocolIncus) getEvents(allProjects bool, eventTypes []string) (*Event // Send the message to all handlers r.eventListenersLock.Lock() for _, listener := range r.eventListeners[listener.projectName] { - listener.send(event) + listener.targetsLock.Lock() + for _, target := range listener.targets { + if target.types != nil && !slices.Contains(target.types, event.Type) { + continue + } + + go target.function(event) + } + + listener.targetsLock.Unlock() } r.eventListenersLock.Unlock() From c650d8bebad5cbc5b0f667221f86b0b1563039bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ren=C3=A9=20Jochum?= Date: Fri, 7 Aug 2026 21:14:55 +0200 Subject: [PATCH 2/4] client: Add EventListener.AddChannel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Handlers are called concurrently so they can observe events out of order. A channel delivers them one at a time in the order they arrived and lets the reader apply backpressure. A reader that falls too far behind is dropped rather than having events silently skipped. The overflow path cancels a listener without removing it, so Disconnect has to do the removal unconditionally. Signed-off-by: René Jochum --- client/events.go | 119 +++++++++++++++++++++++++++++++++++++++-- client/incus_events.go | 12 +---- 2 files changed, 116 insertions(+), 15 deletions(-) diff --git a/client/events.go b/client/events.go index 86d3da3b0e9..af7d44f588b 100644 --- a/client/events.go +++ b/client/events.go @@ -3,11 +3,15 @@ package incus import ( "context" "errors" + "slices" "sync" "github.com/lxc/incus/v7/shared/api" ) +// eventChannelSize is the buffer size used for channels added without an explicit size. +const eventChannelSize = 1000 + // The EventListener struct is used to interact with an Incus event stream. type EventListener struct { r *ProtocolIncus @@ -18,6 +22,7 @@ type EventListener struct { // projectName stores which project this event listener is associated with (empty for all projects). projectName string targets []*EventTarget + channels []*eventChannel targetsLock sync.Mutex } @@ -27,6 +32,112 @@ type EventTarget struct { types []string } +// The eventChannel struct tracks a channel added through AddChannel. +type eventChannel struct { + ch chan api.Event + types []string +} + +// send passes an event on to the handlers and channels of this listener. +func (e *EventListener) send(event api.Event) { + e.targetsLock.Lock() + defer e.targetsLock.Unlock() + + if e.ctx.Err() != nil { + return + } + + for _, target := range e.targets { + if target.types != nil && !slices.Contains(target.types, event.Type) { + continue + } + + go target.function(event) + } + + for _, entry := range e.channels { + if entry.types != nil && !slices.Contains(entry.types, event.Type) { + continue + } + + select { + case entry.ch <- event: + default: + // Dropping events would leave the reader with a silently incomplete view. + e.err = errors.New("Event channel is too far behind") + e.ctxCancel() + + return + } + } +} + +// AddChannel adds a channel to be sent every matching event, size 0 means use the client's default. +func (e *EventListener) AddChannel(types []string, size int) <-chan api.Event { + if size <= 0 { + size = eventChannelSize + } + + ch := make(chan api.Event, size) + + // Handle locking + e.targetsLock.Lock() + defer e.targetsLock.Unlock() + + // A listener that is already done will never deliver anything. + if e.ctx.Err() != nil { + close(ch) + + return ch + } + + // Close the channels once the listener is done so that readers can range over them. + if e.channels == nil { + context.AfterFunc(e.ctx, e.closeChannels) + } + + e.channels = append(e.channels, &eventChannel{ch: ch, types: types}) + + return ch +} + +// RemoveChannel removes and closes a channel previously added with AddChannel. +func (e *EventListener) RemoveChannel(ch <-chan api.Event) error { + if ch == nil { + return errors.New("A valid channel must be provided") + } + + // Handle locking + e.targetsLock.Lock() + defer e.targetsLock.Unlock() + + // Locate and remove the channel from the list + for i, entry := range e.channels { + if entry.ch == ch { + close(entry.ch) + copy(e.channels[i:], e.channels[i+1:]) + e.channels[len(e.channels)-1] = nil + e.channels = e.channels[:len(e.channels)-1] + + return nil + } + } + + return errors.New("Couldn't find this channel") +} + +// closeChannels closes all remaining channels once the listener is done. +func (e *EventListener) closeChannels() { + e.targetsLock.Lock() + defer e.targetsLock.Unlock() + + for _, entry := range e.channels { + close(entry.ch) + } + + e.channels = nil +} + // AddHandler adds a function to be called whenever an event is received. func (e *EventListener) AddHandler(types []string, function func(api.Event)) (*EventTarget, error) { if function == nil { @@ -78,10 +189,6 @@ func (e *EventListener) Disconnect() { e.r.eventListenersLock.Lock() defer e.r.eventListenersLock.Unlock() - if e.ctx.Err() != nil { - return - } - // Locate and remove it from the global list for i, listener := range e.r.eventListeners[e.projectName] { if listener == e { @@ -92,6 +199,10 @@ func (e *EventListener) Disconnect() { } } + if e.ctx.Err() != nil { + return + } + // Turn off the handler e.err = nil e.ctxCancel() diff --git a/client/incus_events.go b/client/incus_events.go index 5e559d3f7a7..f19fa599bbc 100644 --- a/client/incus_events.go +++ b/client/incus_events.go @@ -5,7 +5,6 @@ import ( "encoding/json" "errors" "net/url" - "slices" "strings" "time" @@ -178,16 +177,7 @@ func (r *ProtocolIncus) getEvents(allProjects bool, eventTypes []string) (*Event // Send the message to all handlers r.eventListenersLock.Lock() for _, listener := range r.eventListeners[listener.projectName] { - listener.targetsLock.Lock() - for _, target := range listener.targets { - if target.types != nil && !slices.Contains(target.types, event.Type) { - continue - } - - go target.function(event) - } - - listener.targetsLock.Unlock() + listener.send(event) } r.eventListenersLock.Unlock() From c32ca6ad362ebfbf26cc3c005cd87ffa95007f6c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ren=C3=A9=20Jochum?= Date: Fri, 7 Aug 2026 21:16:14 +0200 Subject: [PATCH 3/4] doc: document AddChannel/RemoveChannel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: René Jochum --- doc/events.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/doc/events.md b/doc/events.md index 5af48e3b01d..2ebc86c885c 100644 --- a/doc/events.md +++ b/doc/events.md @@ -22,8 +22,9 @@ reconnecting, anything being tracked from the event stream should be re-fetched, disconnected are not resent. When using the Go client, handlers registered through `AddHandler` are called concurrently and may therefore observe -events out of order. Calling `SetOrdered` on the listener before adding any handler makes them run one event at a time -and in order instead. +events out of order. Use `AddChannel` instead to receive the events on a channel, which delivers them one at a time and +in the order they arrived, and is closed once the listener ends. Its size argument sets how far behind the reader may +fall before the listener is dropped. ## Event structure From 0efbdf6e75a0a3672a0f5ffa8876705585f1e630 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ren=C3=A9=20Jochum?= Date: Fri, 7 Aug 2026 21:16:29 +0200 Subject: [PATCH 4/4] doc: format events.md MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: René Jochum --- doc/events.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/doc/events.md b/doc/events.md index 2ebc86c885c..dcffaf0a0c3 100644 --- a/doc/events.md +++ b/doc/events.md @@ -99,8 +99,8 @@ type: lifecycle | `image-alias-deleted` | An alias has been deleted for an existing image. | `target`: the original instance. | | `image-alias-renamed` | The alias for an existing image has been renamed. | `old_name`: the previous name. | | `image-alias-updated` | The configuration for an image alias has changed. | `target`: the original instance. | -| `instance-agent-started` | The instance agent has connected to the host. | | -| `instance-agent-stopped` | The instance agent has disconnected from the host. | | +| `instance-agent-started` | The instance agent has connected to the host. | | +| `instance-agent-stopped` | The instance agent has disconnected from the host. | | | `image-created` | A new image has been added to the image store. | `type`: `container` or `vm`. | | `image-deleted` | The image has been deleted from the image store. | | | `image-refreshed` | The local image copy has updated to the current source image version. | |