issue 1840/S6: emit through the same EventSink as the upgrade parts, so one connection serves both

This commit is contained in:
2026-10-22 14:32:10 +02:00
parent e4fb4cc793
commit 79ce72a7b7
2 changed files with 12 additions and 11 deletions
+6 -9
View File
@@ -77,10 +77,6 @@ type Store interface {
Watch(key string, handle func(stdio.StateChange) error) error
}
// Emitter emits one event with its body; id is the event's id, the bus's dedup id, so emitting it again after a
// failure the bus did not report is dropped.
type Emitter func(event string, body any, id string) error
// Clock is the time as the keeper sees it.
type Clock interface {
Now() time.Time
@@ -154,7 +150,9 @@ type Keeper struct {
Config KeeperConfig
Machine *Machine
State func(name string) Store
Emit Emitter
// Events emits the keeper's events; package.changed carries the id `installed.<node>.<revision>`, the bus's
// dedup id, so the same change said twice is dropped.
Events EventSink
Clock Clock
Watcher Watcher
// Locked says pacman's lock is held now: a released lock followed at once by another transaction is read
@@ -334,7 +332,8 @@ func (k *Keeper) refreshInstalled(at time.Time) *PackageChanged {
return nil
}
ev := &PackageChanged{Node: k.Node, At: at.UTC(), Added: c.Added, Removed: c.Removed, Changed: c.Changed}
if err := k.Emit(EventPackageChanged, *ev, fmt.Sprintf("%s.%s.%d", StateInstalled, k.Node, rev)); err != nil {
id := fmt.Sprintf("%s.%s.%d", StateInstalled, k.Node, rev)
if err := k.Events.Emit(EventPackageChanged, *ev, stdio.EmitOptions{MessageID: id}); err != nil {
k.logf("[pacman] keeper: emitting %s for installed.%s at revision %d: %v", EventPackageChanged, k.Node, rev, err)
}
return ev
@@ -521,8 +520,6 @@ func runKeeper(ctx context.Context, account process.Account, k *Keeper) error {
k.Node = account.Node
}
k.State = func(name string) Store { return processStore{conn.State(name)} }
k.Emit = func(event string, body any, id string) error {
return conn.Emit(event, body, stdio.EmitOptions{MessageID: id})
}
k.Events = conn
return k.Run(ctx)
}
@@ -202,9 +202,13 @@ type emittedEvent struct {
id string
}
func (e *emitted) Emit(name string, body any, id string) error {
func (e *emitted) Emit(name string, body any, opts ...stdio.EmitOptions) error {
e.mu.Lock()
defer e.mu.Unlock()
id := ""
for _, o := range opts {
id = o.MessageID
}
e.events = append(e.events, emittedEvent{name, body, id})
return nil
}
@@ -318,7 +322,7 @@ func newTestKeeper(db *pacmanDB) (*Keeper, *fakeStates, *emitted, *fakeClock, fa
states, ev, clock := &fakeStates{}, &emitted{}, &fakeClock{now: t0}
w := fakeWatcher{make(chan Signal)}
m := machine(db.runner(), 1000)
k := &Keeper{Node: "node-a", Machine: m, State: states.State, Emit: ev.Emit, Clock: clock, Watcher: w,
k := &Keeper{Node: "node-a", Machine: m, State: states.State, Events: ev, Clock: clock, Watcher: w,
Logf: func(string, ...any) {}}
return k, states, ev, clock, w
}