From 07b9d32374b86678e2e444abf4bd307bff732702 Mon Sep 17 00:00:00 2001 From: ephraimduncan Date: Thu, 5 Nov 2026 20:14:34 +0000 Subject: [PATCH 001/163] refactor(events): give each bus event its own type Each event of the bus is now a type of the events package, and the type is the topic. A publisher cannot send the data of a different topic, and a consumer does a type switch or reads a typed channel. events.Event{Type, Data any}, events.As, and the JSON convert of internal/api are removed. events.PoolStats is the one pool.stats type for the scheduler, the SSE stream, and the OpenAPI schema. events.HostStats is store.ServerStat with the name of the SSE schema. The copies in internal/api and the scheduler are removed. The OpenAPI spec does not change. The bus has one delivery rule for each kind of subscriber: - events.Subscribe[E] gives each event of one type and never loses one. When the subscriber is slow, the events wait in a queue without a size limit. notify, github, and fleet use it. - Bus.Watch gives all events to a client that can read the state again. When more than buffer events wait, the channel closes. The SSE stream then ends, and the browser connects again and reads all its queries. Before, a full buffer of 64 or 256 events lost settings changes and job failures for github and notify without a record. A browser lost events and kept old data until it connected again. The image row from the API and the image build from the fleet are two types, ImageState and ImageBuildState. The fleet provisions the images only for ImageState, as before. The SSE stream sends the two types as image.state. --- cmd/runplane/services.go | 2 +- cmd/runplane/services_test.go | 9 +- internal/api/api_test.go | 2 +- internal/api/events.go | 126 ++++++----------- internal/api/github_test.go | 4 +- internal/api/images.go | 4 +- internal/api/maintenance.go | 2 +- internal/api/servers.go | 2 +- internal/api/servers_test.go | 2 +- internal/api/settings.go | 2 +- internal/events/events.go | 208 +++++++++++++++++++++-------- internal/events/events_test.go | 76 ++++++++--- internal/fleet/fleet.go | 28 ++-- internal/fleet/helpers_test.go | 2 +- internal/fleet/provision.go | 4 +- internal/fleet/servers_test.go | 20 ++- internal/fleet/stats.go | 2 +- internal/github/github.go | 11 +- internal/github/github_test.go | 9 +- internal/github/helpers_test.go | 10 +- internal/github/releases.go | 2 +- internal/github/releases_test.go | 7 +- internal/notify/dispatch.go | 76 +++++------ internal/notify/dispatch_test.go | 20 +-- internal/scheduler/cleanup.go | 4 +- internal/scheduler/cleanup_test.go | 7 +- internal/scheduler/jobs.go | 8 +- internal/scheduler/resize.go | 2 +- internal/scheduler/scale.go | 12 +- internal/scheduler/scale_test.go | 15 +-- internal/scheduler/scheduler.go | 20 +-- 31 files changed, 370 insertions(+), 328 deletions(-) diff --git a/cmd/runplane/services.go b/cmd/runplane/services.go index 107c173..1fdcbd8 100644 --- a/cmd/runplane/services.go +++ b/cmd/runplane/services.go @@ -163,7 +163,7 @@ func (b *backups) save(ctx context.Context, row store.Backup) { } func (b *backups) publish(row store.Backup) { - b.bus.Publish(events.Event{Type: events.BackupState, Data: row}) + b.bus.Publish(events.BackupState(row)) } // dropRows deletes the rows of the backup files that the retention removed. diff --git a/cmd/runplane/services_test.go b/cmd/runplane/services_test.go index 1dc171c..37fca12 100644 --- a/cmd/runplane/services_test.go +++ b/cmd/runplane/services_test.go @@ -100,16 +100,13 @@ func TestBackupStateEvents(t *testing.T) { } } bus := &events.Bus{} - states, stop := bus.Subscribe(16) + states := events.Subscribe[events.BackupState](t.Context(), bus) notified := 0 b := &backups{st: st, box: box, bus: bus, dir: tt.dir(t), notify: func(notify.Event) { notified++ }, log: slog.New(slog.DiscardHandler)} b.Run(ctx) - stop() var seen []string - for e := range states { - if row, ok := e.Data.(store.Backup); ok && e.Type == events.BackupState { - seen = append(seen, row.Status) - } + for len(states) > 0 { + seen = append(seen, (<-states).Status) } if !slices.Equal(seen, tt.want) { t.Errorf("backup.state events %v, want %v", seen, tt.want) diff --git a/internal/api/api_test.go b/internal/api/api_test.go index a0e6132..9d6a415 100644 --- a/internal/api/api_test.go +++ b/internal/api/api_test.go @@ -596,7 +596,7 @@ func TestEventsStreamConvertsStoreModels(t *testing.T) { pool := store.Pool{ID: ids.New(), Name: "Default", Label: "runplane", Arch: "x64", VCPU: 2, MemoryMB: 4096, DiskGB: 20, MaxRunners: 1, Enabled: true, CreatedAt: now, UpdatedAt: now} must(t, e.st.CreatePool(context.Background(), pool)) sb := store.Sandbox{ID: "sb1", PoolID: pool.ID, ServerID: "srv", State: sandbox.StateBusy, VCPU: 2, MemoryMB: 4096, CreatedAt: now} - e.publish(events.Event{Type: events.SettingsChanged, Data: store.Settings{}}, events.Event{Type: events.SandboxState, Data: &sb}) + e.publish(events.SettingsChanged{}, events.SandboxState(sb)) ev := recv(t, ch) if ev.name != "sandbox.state" { diff --git a/internal/api/events.go b/internal/api/events.go index a5bad0f..a964a0b 100644 --- a/internal/api/events.go +++ b/internal/api/events.go @@ -2,8 +2,8 @@ package api import ( "context" - "encoding/json" "errors" + "fmt" "time" "github.com/danielgtaylor/huma/v2/sse" @@ -13,45 +13,31 @@ import ( "github.com/runplane/runplane/internal/store" ) -type PoolStats struct { - PoolID string `json:"pool_id"` - Idle int `json:"idle"` - Busy int `json:"busy"` - Starting int `json:"starting"` - Queued int `json:"queued"` - Assigned int `json:"assigned"` -} - -type HostStats struct { - ServerID string `json:"server_id"` - At time.Time `json:"at"` - CPUPercent float64 `json:"cpu_percent"` - MemoryPercent float64 `json:"memory_percent"` - DiskPercent float64 `json:"disk_percent"` -} - const pingEvery = 15 * time.Second +// registerEvents names the SSE events. sse.Sender finds the name of an event +// from the type of its data, so each name has its own data type. func (h *Handler) registerEvents() { op := operation(auth.ActionRead, "GET /api/events", "stream-events", "Stream live events (SSE)", describe( "Server-Sent Events. Each event has the API object of its name as data. A comment line \": ping\" comes each 15 seconds. "+ "Only admins get backup.state.")) sse.Register(h.api, op, map[string]any{ - events.SandboxState: Sandbox{}, - events.JobState: JobSummary{}, - events.PoolStats: PoolStats{}, - events.ServerState: Server{}, - events.HostStats: HostStats{}, - events.RunnerVersion: RunnerVersion{}, - events.ImageState: Image{}, - events.GitHubState: GitHubStatus{}, - events.BackupState: Backup{}, + "sandbox.state": Sandbox{}, + "job.state": JobSummary{}, + "pool.stats": events.PoolStats{}, + "server.state": Server{}, + "host.stats": events.HostStats{}, + "runner.version": RunnerVersion{}, + "image.state": Image{}, + "github.state": GitHubStatus{}, + "backup.state": Backup{}, }, h.streamEvents) } +// streamEvents ends when the browser reads the events too slowly. The +// browser then connects again and reads all its queries again. func (h *Handler) streamEvents(ctx context.Context, _ *struct{}, send sse.Sender) { - ch, cancel := h.d.Bus.Subscribe(256) - defer cancel() + ch := h.d.Bus.Watch(ctx, 256) ping := time.NewTicker(pingEvery) defer ping.Stop() for { @@ -66,11 +52,12 @@ func (h *Handler) streamEvents(ctx context.Context, _ *struct{}, send sse.Sender } case e, ok := <-ch: if !ok { + h.d.Log.Info("stop a slow event stream") return } data, err := h.eventData(ctx, e) if err != nil { - h.d.Log.Warn("convert event", "type", e.Type, "err", err) + h.d.Log.Warn("convert event", "type", fmt.Sprintf("%T", e), "err", err) continue } if data != nil && send.Data(data) != nil { @@ -80,80 +67,43 @@ func (h *Handler) streamEvents(ctx context.Context, _ *struct{}, send sse.Sender } } -// eventData converts the data of a bus event into the API object of the -// event. Publishers can send the API object, the store model, or a value -// with the same JSON fields as the API object. It returns nil for events -// that the browser does not get. +// eventData returns the API object of a bus event, or nil for an event that +// the browser does not get. func (h *Handler) eventData(ctx context.Context, e events.Event) (any, error) { - switch e.Type { + switch e := e.(type) { case events.SandboxState: - if b, ok := events.As[store.Sandbox](e.Data); ok { - return first(h.sandboxes(ctx, []store.Sandbox{b})) - } - return convert[Sandbox](e.Data) + return first(h.sandboxes(ctx, []store.Sandbox{store.Sandbox(e)})) case events.JobState: - if j, ok := events.As[store.Job](e.Data); ok { - return first(h.jobSummaries(ctx, []store.Job{j})) - } - return convert[JobSummary](e.Data) - case events.PoolStats: - return convert[PoolStats](e.Data) + return first(h.jobSummaries(ctx, []store.Job{store.Job(e)})) + case events.PoolStats, events.HostStats: + return e, nil case events.ServerState: - if v, ok := events.As[store.Server](e.Data); ok { - return first(h.servers(ctx, []store.Server{v})) - } - return convert[Server](e.Data) - case events.HostStats: - if st, ok := events.As[store.ServerStat](e.Data); ok { - return HostStats(st), nil - } - return convert[HostStats](e.Data) + return first(h.servers(ctx, []store.Server{store.Server(e)})) case events.RunnerVersion: - if b, ok := events.As[store.RunnerBundle](e.Data); ok { - return runnerVersion(b, time.Now()), nil - } - return convert[RunnerVersion](e.Data) + return runnerVersion(store.RunnerBundle(e), time.Now()), nil case events.ImageState: - id := "" - if img, ok := events.As[store.Image](e.Data); ok { - id = img.ID - } else if b, ok := events.As[store.ImageBuild](e.Data); ok { - id = b.ImageID - } else { - return convert[Image](e.Data) - } - // Read the image again for its current builds. - img, err := h.d.Store.GetImage(ctx, id) - if errors.Is(err, store.ErrNotFound) { - return nil, nil - } - if err != nil { - return nil, err - } - return first(h.images(ctx, []store.Image{img})) + return h.imageEvent(ctx, e.ID) + case events.ImageBuildState: + return h.imageEvent(ctx, e.ImageID) case events.GitHubState: return h.gitHubStatus(ctx) case events.BackupState: if !auth.CanDo(currentUser(ctx).Role, auth.ActionAdmin) { return nil, nil } - if b, ok := events.As[store.Backup](e.Data); ok { - return h.backupOut(b), nil - } - return convert[Backup](e.Data) + return h.backupOut(store.Backup(e)), nil } return nil, nil } -// convert returns v as a T, or through JSON for another type with the same fields. -func convert[T any](v any) (T, error) { - if x, ok := events.As[T](v); ok { - return x, nil +// imageEvent reads the image again for its current builds. +func (h *Handler) imageEvent(ctx context.Context, id string) (any, error) { + img, err := h.d.Store.GetImage(ctx, id) + if errors.Is(err, store.ErrNotFound) { + return nil, nil } - var out T - b, err := json.Marshal(v) - if err == nil { - err = json.Unmarshal(b, &out) + if err != nil { + return nil, err } - return out, err + return first(h.images(ctx, []store.Image{img})) } diff --git a/internal/api/github_test.go b/internal/api/github_test.go index c2688ed..49cdb6c 100644 --- a/internal/api/github_test.go +++ b/internal/api/github_test.go @@ -47,8 +47,8 @@ func TestGitHubAndBackupEvents(t *testing.T) { e := newEnv(t) member := e.login("bob@example.com", auth.RoleMember).events(t) admin := e.login("ada@example.com", auth.RoleAdmin).events(t) - backup := store.Backup{ID: "bk1", CreatedAt: time.Now(), Status: "running"} - e.publish(events.Event{Type: events.BackupState, Data: backup}, events.Event{Type: events.GitHubState}) + backup := events.BackupState{ID: "bk1", CreatedAt: time.Now(), Status: "running"} + e.publish(backup, events.GitHubState{}) for seen := 0; seen < 2; { ev := recv(t, member) diff --git a/internal/api/images.go b/internal/api/images.go index 5fa428f..2f96a56 100644 --- a/internal/api/images.go +++ b/internal/api/images.go @@ -105,7 +105,7 @@ func (h *Handler) createImage(ctx context.Context, in *body[CreateImageRequest]) return nil, imageErr(err) } h.audit(ctx, "image.create", img.ID, map[string]any{"name": img.Name, "ref": img.Ref}) - h.d.Bus.Publish(events.Event{Type: events.ImageState, Data: img}) + h.d.Bus.Publish(events.ImageState(img)) h.refresh(ctx, img.ID) return h.imageOutput(ctx, img.ID) } @@ -151,7 +151,7 @@ func (h *Handler) updateImage(ctx context.Context, in *struct { h.poolsChanged(ctx) } h.audit(ctx, "image.update", img.ID, map[string]any{"name": img.Name, "ref": img.Ref, "is_default": b.IsDefault}) - h.d.Bus.Publish(events.Event{Type: events.ImageState, Data: img}) + h.d.Bus.Publish(events.ImageState(img)) if refChanged { h.refresh(ctx, img.ID) } diff --git a/internal/api/maintenance.go b/internal/api/maintenance.go index 6092435..c127f6c 100644 --- a/internal/api/maintenance.go +++ b/internal/api/maintenance.go @@ -59,7 +59,7 @@ func (h *Handler) createBackup(ctx context.Context, _ *struct{}) (*body[Backup], return nil, notFound(err, "The new backup is gone.") } h.audit(ctx, "backup.create", b.ID, nil) - h.d.Bus.Publish(events.Event{Type: events.BackupState, Data: b}) + h.d.Bus.Publish(events.BackupState(b)) return &body[Backup]{h.backupOut(b)}, nil } diff --git a/internal/api/servers.go b/internal/api/servers.go index 2263db4..a84c4c0 100644 --- a/internal/api/servers.go +++ b/internal/api/servers.go @@ -278,7 +278,7 @@ func (h *Handler) updateServer(ctx context.Context, in *struct { if v, err = h.d.Store.GetServer(ctx, v.ID); err != nil { return nil, notFound(err, "No server has this ID.") } - h.d.Bus.Publish(events.Event{Type: events.ServerState, Data: v}) + h.d.Bus.Publish(events.ServerState(v)) changes := map[string][2]any{} if v.Name != old.Name { changes["name"] = [2]any{old.Name, v.Name} diff --git a/internal/api/servers_test.go b/internal/api/servers_test.go index d13afae..038094e 100644 --- a/internal/api/servers_test.go +++ b/internal/api/servers_test.go @@ -248,7 +248,7 @@ func TestServerStateEventIsTheServerObject(t *testing.T) { v, err := e.st.GetServer(ctx, "s1") must(t, err) - e.publish(events.Event{Type: events.ServerState, Data: v}) + e.publish(events.ServerState(v)) ev := recv(t, ch) var got Server must(t, json.Unmarshal([]byte(ev.data), &got)) diff --git a/internal/api/settings.go b/internal/api/settings.go index 87762ac..bd9fcd6 100644 --- a/internal/api/settings.go +++ b/internal/api/settings.go @@ -156,7 +156,7 @@ func (h *Handler) updateSettings(ctx context.Context, in *body[SettingsPatch]) ( } if changes := fieldChanges(old, next); len(changes) > 0 { h.audit(ctx, "settings.update", "", settingsDetail(old, next, changes)) - h.d.Bus.Publish(events.Event{Type: events.SettingsChanged, Data: next}) + h.d.Bus.Publish(events.SettingsChanged(next)) if h.d.SettingsHook != nil { h.d.SettingsHook.SettingsChanged(context.WithoutCancel(ctx), old, next) } diff --git a/internal/events/events.go b/internal/events/events.go index f7f40ab..330685c 100644 --- a/internal/events/events.go +++ b/internal/events/events.go @@ -1,78 +1,182 @@ -// Package events is the in-process publish and subscribe bus. The API sends -// the events to the browser over SSE. +// Package events is the in-process bus. Each event type of this package is a +// topic and carries its own data, so a publisher cannot send the data of +// another topic. +// +// The bus has one delivery rule for each kind of subscriber. Subscribe never +// loses an event, for the packages that must see each event: notify must see +// each failed job, and github each settings change. Watch ends when its client +// reads too slowly. The SSE stream of the API uses Watch, and the browser then +// connects again and reads the state from the API. package events -import "sync" - -// Event types. All except SettingsChanged are SSE event names. -const ( - SandboxState = "sandbox.state" - JobState = "job.state" - PoolStats = "pool.stats" - ServerState = "server.state" - HostStats = "host.stats" - RunnerVersion = "runner.version" - ImageState = "image.state" - // GitHubState says that the GitHub status changed: the app, the - // installations, or the scale set listeners. It has no data. - GitHubState = "github.state" - BackupState = "backup.state" - - // SettingsChanged is internal: packages that cache settings read them again. - SettingsChanged = "settings.changed" +import ( + "context" + "sync" + + "github.com/runplane/runplane/internal/store" ) -type Event struct { - Type string - Data any -} +// Event is an event of the bus. Only the types of this package are events. +type Event interface{ event() } -// As returns the data of an event as a T, also when the data is a *T. -func As[T any](data any) (T, bool) { - switch x := data.(type) { - case T: - return x, true - case *T: - if x != nil { - return *x, true - } - } - var zero T - return zero, false +// Each of these events has a row after its change. +type ( + SandboxState store.Sandbox + JobState store.Job + ServerState store.Server + RunnerVersion store.RunnerBundle + BackupState store.Backup +) + +// HostStats is a new host sample of a server. The browser gets it as it is. +type HostStats store.ServerStat + +// ImageState is an image row that the API added or changed. The fleet makes +// the images ready again. +type ImageState store.Image + +// ImageBuildState is the build of an image on a server after a change. +type ImageBuildState store.ImageBuild + +// PoolStats counts the runners and the waiting jobs of a pool. The browser +// gets it as it is. +type PoolStats struct { + PoolID string `json:"pool_id"` + Idle int `json:"idle"` + Busy int `json:"busy"` + Starting int `json:"starting"` + Queued int `json:"queued"` + Assigned int `json:"assigned"` } -// Bus delivers each event to all subscribers. The zero value is ready to use. +// GitHubState says that the GitHub status changed: the app, the +// installations, or the scale set listeners. +type GitHubState struct{} + +// SettingsChanged has the new settings. The browser does not get it. +type SettingsChanged store.Settings + +func (SandboxState) event() {} +func (JobState) event() {} +func (ServerState) event() {} +func (HostStats) event() {} +func (RunnerVersion) event() {} +func (BackupState) event() {} +func (ImageState) event() {} +func (ImageBuildState) event() {} +func (PoolStats) event() {} +func (GitHubState) event() {} +func (SettingsChanged) event() {} + +// Bus gives each event to its subscribers. The zero value is ready to use. type Bus struct { - mu sync.RWMutex - subs map[chan Event]struct{} + mu sync.RWMutex + sinks map[*sink]struct{} } -// Publish never blocks. A subscriber with a full buffer loses the event. +// sink takes the events of one subscriber. Its put never waits. +type sink struct{ put func(Event) } + +// Publish gives e to each subscriber. It never waits for a subscriber. func (b *Bus) Publish(e Event) { b.mu.RLock() defer b.mu.RUnlock() - for ch := range b.subs { + for s := range b.sinks { + s.put(e) + } +} + +// Subscribe returns each event of type E until ctx ends, in the order of +// Publish. It never loses an event: when the subscriber is slow, its events +// wait in a queue without a size limit. The channel does not close. +func Subscribe[E Event](ctx context.Context, b *Bus) <-chan E { + q := &queue[E]{ch: make(chan E, queueRoom), done: ctx.Done()} + b.add(ctx, func(e Event) { + if x, ok := e.(E); ok { + q.push(x) + } + }) + return q.ch +} + +// Watch returns all events until ctx ends, for a client that can read the +// state again. When more than buffer events wait for the client, the client +// loses the events and the channel closes, so that the client knows. +func (b *Bus) Watch(ctx context.Context, buffer int) <-chan Event { + ch := make(chan Event, buffer) + var mu sync.Mutex + closed := false + b.add(ctx, func(e Event) { + mu.Lock() + defer mu.Unlock() + if closed { + return + } select { case ch <- e: default: + closed = true + close(ch) } - } + }) + return ch } -// Subscribe returns a channel with room for buffer events. Cancel stops the -// delivery and closes the channel; it is safe to call more than one time. -func (b *Bus) Subscribe(buffer int) (<-chan Event, func()) { - ch := make(chan Event, buffer) +func (b *Bus) add(ctx context.Context, put func(Event)) { + s := &sink{put} b.mu.Lock() - if b.subs == nil { - b.subs = make(map[chan Event]struct{}) + if b.sinks == nil { + b.sinks = map[*sink]struct{}{} } - b.subs[ch] = struct{}{} + b.sinks[s] = struct{}{} b.mu.Unlock() - return ch, sync.OnceFunc(func() { + context.AfterFunc(ctx, func() { b.mu.Lock() - delete(b.subs, ch) + delete(b.sinks, s) b.mu.Unlock() - close(ch) }) } + +// queueRoom is the room of a Subscribe channel. More events wait in the queue. +const queueRoom = 64 + +type queue[E any] struct { + ch chan E + done <-chan struct{} + + mu sync.Mutex + wait []E // the events after a full ch, in order +} + +func (q *queue[E]) push(x E) { + q.mu.Lock() + defer q.mu.Unlock() + if len(q.wait) == 0 { + select { + case q.ch <- x: + return + default: + go q.drain() + } + } + q.wait = append(q.wait, x) +} + +// drain moves the waiting events to ch. One drain runs while wait has +// events: push starts it for the first one, and it stops when wait is empty. +func (q *queue[E]) drain() { + q.mu.Lock() + for len(q.wait) > 0 { + x := q.wait[0] + q.mu.Unlock() + select { + case q.ch <- x: + case <-q.done: + return + } + q.mu.Lock() + q.wait = q.wait[1:] + } + q.wait = nil + q.mu.Unlock() +} diff --git a/internal/events/events_test.go b/internal/events/events_test.go index a480a69..ea26f4b 100644 --- a/internal/events/events_test.go +++ b/internal/events/events_test.go @@ -1,6 +1,7 @@ package events import ( + "context" "sync" "testing" "time" @@ -8,15 +9,17 @@ import ( func TestSlowSubscriberDoesNotBlock(t *testing.T) { var b Bus - slow, cancelSlow := b.Subscribe(1) - defer cancelSlow() - fast, cancelFast := b.Subscribe(10) - defer cancelFast() + ctx := t.Context() + n := 3 * queueRoom + stats := Subscribe[PoolStats](ctx, &b) + slow := b.Watch(ctx, 1) + fast := b.Watch(ctx, 2*n) done := make(chan struct{}) go func() { - for i := range 5 { - b.Publish(Event{Type: JobState, Data: i}) + for i := range n { + b.Publish(PoolStats{Idle: i}) + b.Publish(GitHubState{}) } close(done) }() @@ -26,32 +29,61 @@ func TestSlowSubscriberDoesNotBlock(t *testing.T) { t.Fatal("Publish blocked on a full subscriber") } - if e := <-slow; e.Data != 0 { - t.Errorf("slow got %v, want the first event", e.Data) + for i := range n { + select { + case p := <-stats: + if p.Idle != i { + t.Fatalf("pool.stats event %d has idle %d, want the events in order", i, p.Idle) + } + case <-time.After(5 * time.Second): + t.Fatalf("got %d of %d pool.stats events", i, n) + } } select { - case e := <-slow: - t.Errorf("slow got %v, want the other events dropped", e.Data) + case p := <-stats: + t.Errorf("one more pool.stats event %+v, want none", p) default: } - for i := range 5 { - if e := <-fast; e.Data != i || e.Type != JobState { - t.Fatalf("fast event %d = %+v", i, e) + + if e := <-slow; e != (PoolStats{Idle: 0}) { + t.Errorf("slow watcher got %+v, want the first event", e) + } + select { + case e, ok := <-slow: + if ok { + t.Errorf("slow watcher got %+v, want its channel closed after it lost events", e) + } + case <-time.After(5 * time.Second): + t.Error("the channel of the slow watcher is open after it lost events") + } + for i := range 2 * n { + if _, ok := <-fast; !ok { + t.Fatalf("fast watcher closed after %d events", i) } } } -func TestCancel(t *testing.T) { +func TestSubscriptionEndsWithContext(t *testing.T) { var b Bus - ch, cancel := b.Subscribe(4) + ctx, cancel := context.WithCancel(t.Context()) + Subscribe[JobState](ctx, &b) + b.Watch(ctx, 4) cancel() - cancel() // a second call is safe - b.Publish(Event{Type: SandboxState}) - if _, ok := <-ch; ok { - t.Fatal("the channel is open after cancel") + deadline := time.Now().Add(5 * time.Second) + for subscribers(&b) > 0 { + if time.Now().After(deadline) { + t.Fatalf("%d subscribers after the end of their context", subscribers(&b)) + } + time.Sleep(time.Millisecond) } } +func subscribers(b *Bus) int { + b.mu.RLock() + defer b.mu.RUnlock() + return len(b.sinks) +} + func TestPublishWhileCancel(t *testing.T) { var b Bus var wg sync.WaitGroup @@ -62,12 +94,14 @@ func TestPublishWhileCancel(t *testing.T) { case <-stop: return default: - b.Publish(Event{Type: HostStats}) + b.Publish(HostStats{}) } } }) for range 200 { - _, cancel := b.Subscribe(0) + ctx, cancel := context.WithCancel(t.Context()) + b.Watch(ctx, 0) + Subscribe[HostStats](ctx, &b) cancel() } close(stop) diff --git a/internal/fleet/fleet.go b/internal/fleet/fleet.go index 0aa0331..914f006 100644 --- a/internal/fleet/fleet.go +++ b/internal/fleet/fleet.go @@ -194,8 +194,9 @@ func (f *Fleet) Run(ctx context.Context) error { if err := f.startAgents(ctx); err != nil { return err } - bus, cancel := f.bus.Subscribe(64) - defer cancel() + // The API publishes the server and image rows that it changes. + servers := events.Subscribe[events.ServerState](ctx, f.bus) + images := events.Subscribe[events.ImageState](ctx, f.bus) f.provision(ctx) check := time.NewTicker(syncEvery) @@ -211,8 +212,10 @@ func (f *Fleet) Run(ctx context.Context) error { case <-ctx.Done(): f.flushStats(context.WithoutCancel(ctx)) return nil - case e := <-bus: - f.handleBus(e) + case v := <-servers: + f.serverChanged(store.Server(v)) + case <-images: + f.kick() case <-check.C: f.syncServers(ctx) case <-flush.C: @@ -363,22 +366,7 @@ func (f *Fleet) publishServer(ctx context.Context, id string) { f.log.Debug("read server for its event", "server_id", id, "err", err) return } - f.bus.Publish(events.Event{Type: events.ServerState, Data: v}) -} - -// handleBus follows the changes that the API makes. -func (f *Fleet) handleBus(e events.Event) { - switch e.Type { - case events.ServerState: - if v, ok := events.As[store.Server](e.Data); ok { - f.serverChanged(v) - } - case events.ImageState: - // The API sends the image row after a new image or a new default. - if _, ok := events.As[store.Image](e.Data); ok { - f.kick() - } - } + f.bus.Publish(events.ServerState(v)) } // serverChanged takes the name and the status of a server row for its diff --git a/internal/fleet/helpers_test.go b/internal/fleet/helpers_test.go index 5178d81..130e921 100644 --- a/internal/fleet/helpers_test.go +++ b/internal/fleet/helpers_test.go @@ -201,7 +201,7 @@ func recv[T any](t *testing.T, ch <-chan T) T { } // waitEvent reads ch until an event matches. -func waitEvent(t *testing.T, ch <-chan events.Event, match func(events.Event) bool) { +func waitEvent[E any](t *testing.T, ch <-chan E, match func(E) bool) { t.Helper() for { if match(recv(t, ch)) { diff --git a/internal/fleet/provision.go b/internal/fleet/provision.go index f264774..0e821e9 100644 --- a/internal/fleet/provision.go +++ b/internal/fleet/provision.go @@ -277,7 +277,7 @@ func (f *Fleet) saveBuild(ctx context.Context, b *store.ImageBuild, status, msg f.log.Warn("record image build", "image", b.ImageID, "server_id", b.ServerID, "err", err) return } - f.bus.Publish(events.Event{Type: events.ImageState, Data: *b}) + f.bus.Publish(events.ImageBuildState(*b)) } // ensureRunner makes the first version in versions that works ready on the @@ -334,7 +334,7 @@ func (f *Fleet) saveBundle(ctx context.Context, b store.RunnerBundle) { if err != nil || after.Status == before.Status && after.Error == before.Error { return } - f.bus.Publish(events.Event{Type: events.RunnerVersion, Data: after}) + f.bus.Publish(events.RunnerVersion(after)) } // EnsureLatest makes the runner bundle of version ready on each online diff --git a/internal/fleet/servers_test.go b/internal/fleet/servers_test.go index 33cca95..40bb5b3 100644 --- a/internal/fleet/servers_test.go +++ b/internal/fleet/servers_test.go @@ -25,8 +25,7 @@ func TestAgentUnreachableAfterOneMinute(t *testing.T) { h := newHarness(t, newFakeNode("arm64")) h.run() a, id := h.connectAgent("build-1", newFakeNode("x64")) - bus, cancel := h.bus.Subscribe(4096) - defer cancel() + servers := events.Subscribe[events.ServerState](ctx, h.bus) a.stop() waitFor(t, "the end of the session", func() bool { return h.f.Driver(id) == nil }) @@ -41,9 +40,8 @@ func TestAgentUnreachableAfterOneMinute(t *testing.T) { if got := h.server(id).Status; got != store.ServerUnreachable { t.Fatalf("status %q 61 seconds after the disconnect, want unreachable", got) } - waitEvent(t, bus, func(e events.Event) bool { - v, ok := e.Data.(store.Server) - return e.Type == events.ServerState && ok && v.ID == id && v.Status == store.ServerUnreachable + waitEvent(t, servers, func(v events.ServerState) bool { + return v.ID == id && v.Status == store.ServerUnreachable }) if e := recv(t, h.notes); e.Type != notify.ServerUnreachable || e.Data["server_id"] != id || len(h.notes) != 0 { t.Errorf("notification %+v and %d more, want one server.unreachable for %s", e, len(h.notes), id) @@ -84,7 +82,7 @@ func TestOnlineServers(t *testing.T) { if err := h.st.SetServerStatus(ctx, agentID, status, ""); err != nil { t.Fatal(err) } - h.bus.Publish(events.Event{Type: events.ServerState, Data: h.server(agentID)}) + h.bus.Publish(events.ServerState(h.server(agentID))) } before := h.capacity.Load() setStatus(store.ServerDraining) @@ -113,7 +111,7 @@ func TestConnectedAgentStaysOnline(t *testing.T) { if err := h.st.SetServerStatus(ctx, id, store.ServerUnreachable, "No connection to the agent."); err != nil { t.Fatal(err) } - h.bus.Publish(events.Event{Type: events.ServerState, Data: h.server(id)}) + h.bus.Publish(events.ServerState(h.server(id))) h.f.syncServers(ctx) if got := h.server(id).Status; got != store.ServerOnline { t.Errorf("status %q of a connected agent, want online", got) @@ -255,8 +253,7 @@ func TestHostStats(t *testing.T) { local := newFakeNode("arm64") h := newHarness(t, local) h.run() - bus, cancel := h.bus.Subscribe(4096) - defer cancel() + stats := events.Subscribe[events.HostStats](ctx, h.bus) id := h.localID() at := h.clock.now().Add(-time.Second) // With a domain and no public URL, the links go to the domain. @@ -269,9 +266,8 @@ func TestHostStats(t *testing.T) { sample := func(diskPercent float64) { local.stats <- host.Stats{At: at, CPUPercent: 52, MemoryPercent: 38, DiskPercent: diskPercent, Resources: host.Resources{DiskBytes: 500e9}} at = at.Add(10 * time.Second) - waitEvent(t, bus, func(e events.Event) bool { - st, ok := e.Data.(store.ServerStat) - return e.Type == events.HostStats && ok && st.ServerID == id && st.DiskPercent == diskPercent + waitEvent(t, stats, func(st events.HostStats) bool { + return st.ServerID == id && st.DiskPercent == diskPercent }) } diff --git a/internal/fleet/stats.go b/internal/fleet/stats.go index a54715d..4febf8f 100644 --- a/internal/fleet/stats.go +++ b/internal/fleet/stats.go @@ -26,7 +26,7 @@ func (f *Fleet) addStats(n *node, s host.Stats) { st := store.ServerStat{ ServerID: n.id, At: s.At.UTC(), CPUPercent: s.CPUPercent, MemoryPercent: s.MemoryPercent, DiskPercent: s.DiskPercent, } - f.bus.Publish(events.Event{Type: events.HostStats, Data: st}) + f.bus.Publish(events.HostStats(st)) free := 100 - s.DiskPercent f.mu.Lock() diff --git a/internal/github/github.go b/internal/github/github.go index c2f98bf..d7dd059 100644 --- a/internal/github/github.go +++ b/internal/github/github.go @@ -120,8 +120,7 @@ func New(cfg Config) *Service { // 5 minutes and checks the runner releases each hour. It returns after ctx // ends and all listeners have stopped. func (s *Service) Run(ctx context.Context) { - settings, unsubscribe := s.bus.Subscribe(64) - defer unsubscribe() + settings := events.Subscribe[events.SettingsChanged](ctx, s.bus) var wg sync.WaitGroup wg.Go(func() { s.releaseLoop(ctx) }) @@ -140,10 +139,8 @@ func (s *Service) Run(ctx context.Context) { s.syncAll(ctx) case <-s.poke: s.reconcileNow(ctx) - case e := <-settings: - if e.Type == events.SettingsChanged { - s.reconcileNow(ctx) - } + case <-settings: + s.reconcileNow(ctx) } } } @@ -197,7 +194,7 @@ func (s *Service) anyInstallation(ctx context.Context) (store.Installation, bool // changed tells the browsers that the GitHub status changed. func (s *Service) changed() { - s.bus.Publish(events.Event{Type: events.GitHubState}) + s.bus.Publish(events.GitHubState{}) } // polledChanged tells the browsers about a new poll time, at most once in diff --git a/internal/github/github_test.go b/internal/github/github_test.go index 224fdb1..a6ae0de 100644 --- a/internal/github/github_test.go +++ b/internal/github/github_test.go @@ -43,7 +43,7 @@ func TestRunFollowsSettingsAndPools(t *testing.T) { eventually(t, "one scale set for the private repository", listening(1)) e.st.UpdateSettings(ctx, func(s *store.Settings) error { s.AllowPublicRepos = true; return nil }) - e.bus.Publish(events.Event{Type: events.SettingsChanged}) + e.bus.Publish(events.SettingsChanged{}) eventually(t, "a scale set for the public repository too", listening(2)) e.pool("p2", "runplane-big", true) @@ -75,8 +75,7 @@ func TestRunFollowsSettingsAndPools(t *testing.T) { func TestGitHubStateEvents(t *testing.T) { e := newEnv(t) e.connect() - published, unsubscribe := e.bus.Subscribe(16) - defer unsubscribe() + published := events.Subscribe[events.GitHubState](t.Context(), e.bus) // Polls publish at most once in 30 seconds. t0 := time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC) for _, step := range []struct { @@ -90,14 +89,14 @@ func TestGitHubStateEvents(t *testing.T) { {40 * time.Second, 0}, } { e.svc.polledChanged(t0.Add(step.at)) - if got := countEvents(published, events.GitHubState); got != step.want { + if got := countEvents(published); got != step.want { t.Errorf("poll at +%s: %d github.state events, want %d", step.at, got, step.want) } } if err := e.svc.Disconnect(context.Background()); err != nil { t.Fatal(err) } - if countEvents(published, events.GitHubState) == 0 { + if countEvents(published) == 0 { t.Error("no github.state event after the app was removed") } } diff --git a/internal/github/helpers_test.go b/internal/github/helpers_test.go index 6c816d0..52e3fec 100644 --- a/internal/github/helpers_test.go +++ b/internal/github/helpers_test.go @@ -158,15 +158,13 @@ func writeJSON(w http.ResponseWriter, status int, v any) { json.NewEncoder(w).Encode(v) } -// countEvents counts the events of type typ on ch so far. -func countEvents(ch <-chan events.Event, typ string) int { +// countEvents counts the events on ch so far. +func countEvents[E any](ch <-chan E) int { n := 0 for { select { - case e := <-ch: - if e.Type == typ { - n++ - } + case <-ch: + n++ default: return n } diff --git a/internal/github/releases.go b/internal/github/releases.go index 5bb53e6..c614d65 100644 --- a/internal/github/releases.go +++ b/internal/github/releases.go @@ -160,7 +160,7 @@ func (s *Service) setBundleDates(ctx context.Context, version, arch string, rele if berr == nil && sameTime(before.ReleasedAt, after.ReleasedAt) && sameTime(before.EndDate, after.EndDate) { return } - s.bus.Publish(events.Event{Type: events.RunnerVersion, Data: after}) + s.bus.Publish(events.RunnerVersion(after)) } // warnEnds sends one notification for each runner version that new diff --git a/internal/github/releases_test.go b/internal/github/releases_test.go index fe6f9a2..3a1c4b8 100644 --- a/internal/github/releases_test.go +++ b/internal/github/releases_test.go @@ -56,8 +56,7 @@ func TestCheckReleases(t *testing.T) { writeJSON(w, http.StatusNotFound, map[string]any{"message": "Not Found"}) } }) - published, unsubscribe := e.bus.Subscribe(16) - defer unsubscribe() + published := events.Subscribe[events.RunnerVersion](t.Context(), e.bus) if err := e.svc.checkReleases(ctx); err != nil { t.Fatal(err) @@ -79,7 +78,7 @@ func TestCheckReleases(t *testing.T) { t.Errorf("2.330.0 %s end date = %v, want the runtime end %v", arch, old.EndDate, runtimeEnd) } } - if n := countEvents(published, events.RunnerVersion); n != 4 { + if n := countEvents(published); n != 4 { t.Errorf("published %d runner.version events, want 4", n) } sent := e.notifier.sent @@ -91,7 +90,7 @@ func TestCheckReleases(t *testing.T) { if err := e.svc.checkReleases(ctx); err != nil { t.Fatal(err) } - if len(e.notifier.sent) != tt.wantWarns || countEvents(published, events.RunnerVersion) != 0 { + if len(e.notifier.sent) != tt.wantWarns || countEvents(published) != 0 { t.Errorf("the second check sent %d warnings and published events, want %d and none", len(e.notifier.sent), tt.wantWarns) } }) diff --git a/internal/notify/dispatch.go b/internal/notify/dispatch.go index c4dff3e..1627715 100644 --- a/internal/notify/dispatch.go +++ b/internal/notify/dispatch.go @@ -130,8 +130,10 @@ func (d *Dispatcher) admit(e Event) bool { // Run makes alerts from the bus until ctx ends. func (d *Dispatcher) Run(ctx context.Context, bus *events.Bus) { - ch, cancel := bus.Subscribe(256) - defer cancel() + settings := events.Subscribe[events.SettingsChanged](ctx, bus) + jobs := events.Subscribe[events.JobState](ctx, bus) + servers := events.Subscribe[events.ServerState](ctx, bus) + stats := events.Subscribe[events.HostStats](ctx, bus) st, err := d.Store.GetSettings(ctx) if err != nil { d.log().Warn("read settings for notifications", "err", err) @@ -155,51 +157,45 @@ func (d *Dispatcher) Run(ctx context.Context, bus *events.Bus) { delete(d.waiting, id) d.mu.Unlock() d.jobFailed(ctx, id) - case e := <-ch: - d.handle(ctx, e) + case st := <-settings: + d.settingsChanged(st) + case job := <-jobs: + d.jobDone(ctx, job) + case v := <-servers: + d.serverState(ctx, v) + case s := <-stats: + d.hostStats(ctx, s) } } } -func (d *Dispatcher) handle(ctx context.Context, e events.Event) { - switch e.Type { - case events.SettingsChanged: - if st, ok := events.As[store.Settings](e.Data); ok { - d.lock() - d.minFree = st.MinFreeDiskPercent - d.mu.Unlock() - } - case events.JobState: - if job, ok := events.As[store.Job](e.Data); ok && job.State == store.JobCompleted && job.Result == store.ResultFailed { - d.jobDone(ctx, job) - } - case events.ServerState: - v, ok := events.As[store.Server](e.Data) - if !ok { - return - } - switch v.Status { - case store.ServerUnreachable: - lastSeen := d.clock() - if v.LastSeenAt != nil { - lastSeen = *v.LastSeenAt - } - d.goNotify(ctx, func(dash string) Event { return ServerUnreachableEvent(dash, v.ID, v.Name, lastSeen) }) - case store.ServerOnline: - d.lock() - delete(d.down, v.ID) - d.mu.Unlock() - } - case events.HostStats: - if s, ok := events.As[store.ServerStat](e.Data); ok { - d.hostStats(ctx, s) +func (d *Dispatcher) settingsChanged(st events.SettingsChanged) { + d.lock() + d.minFree = st.MinFreeDiskPercent + d.mu.Unlock() +} + +func (d *Dispatcher) serverState(ctx context.Context, v events.ServerState) { + switch v.Status { + case store.ServerUnreachable: + lastSeen := d.clock() + if v.LastSeenAt != nil { + lastSeen = *v.LastSeenAt } + d.goNotify(ctx, func(dash string) Event { return ServerUnreachableEvent(dash, v.ID, v.Name, lastSeen) }) + case store.ServerOnline: + d.lock() + delete(d.down, v.ID) + d.mu.Unlock() } } -// jobDone sends the alert of a failed job when its run data is there, or -// when the wait for the data ends. -func (d *Dispatcher) jobDone(ctx context.Context, job store.Job) { +// jobDone sends the alert of a job that failed when its run data is there, +// or when the wait for the data ends. +func (d *Dispatcher) jobDone(ctx context.Context, job events.JobState) { + if job.State != store.JobCompleted || job.Result != store.ResultFailed { + return + } d.lock() defer d.mu.Unlock() if _, sent := d.jobs[job.ID]; sent { @@ -239,7 +235,7 @@ func (d *Dispatcher) jobFailed(ctx context.Context, id string) { // hostStats sends a disk.low alert when the free disk space of a server // falls below min_free_disk_percent. The next alert waits until the space // rises diskMargin points above the limit and falls again. -func (d *Dispatcher) hostStats(ctx context.Context, s store.ServerStat) { +func (d *Dispatcher) hostStats(ctx context.Context, s events.HostStats) { free := 100 - s.DiskPercent d.lock() minFree := d.minFree diff --git a/internal/notify/dispatch_test.go b/internal/notify/dispatch_test.go index 00063c2..0b670d8 100644 --- a/internal/notify/dispatch_test.go +++ b/internal/notify/dispatch_test.go @@ -153,7 +153,7 @@ func TestDispatcherSendsOneAlertForAFailedJob(t *testing.T) { // The completion comes before the run data: no alert yet. for range 15 { for _, j := range jobs { - bus.Publish(events.Event{Type: events.JobState, Data: j}) + bus.Publish(events.JobState(j)) } time.Sleep(10 * time.Millisecond) } @@ -167,9 +167,9 @@ func TestDispatcherSendsOneAlertForAFailedJob(t *testing.T) { if err != nil { t.Fatal(err) } - publishUntilAlert(t, bus, h, events.Event{Type: events.JobState, Data: &enriched}) + publishUntilAlert(t, bus, h, events.JobState(enriched)) for range 3 { - bus.Publish(events.Event{Type: events.JobState, Data: enriched}) + bus.Publish(events.JobState(enriched)) } time.Sleep(100 * time.Millisecond) got := h.events() @@ -194,7 +194,7 @@ func TestDispatcherSendsAFailedJobWithoutRunData(t *testing.T) { if err != nil { t.Fatal(err) } - publishUntilAlert(t, bus, h, events.Event{Type: events.JobState, Data: j}) + publishUntilAlert(t, bus, h, events.JobState(j)) if got := h.events(); len(got) != 1 || got[0].Data["job_id"] != j.ID { t.Errorf("alerts %+v, want one for %s", got, j.ID) } @@ -211,10 +211,10 @@ func TestDispatcherDiskLow(t *testing.T) { } now := at d.now = func() time.Time { return now } - d.handle(ctx, events.Event{Type: events.SettingsChanged, Data: store.Settings{MinFreeDiskPercent: 15}}) + d.settingsChanged(events.SettingsChanged{MinFreeDiskPercent: 15}) sample := func(after time.Duration, usedPercent float64) { now = at.Add(after) - d.handle(ctx, events.Event{Type: events.HostStats, Data: store.ServerStat{ServerID: "s1", At: now, DiskPercent: usedPercent}}) + d.hostStats(ctx, events.HostStats{ServerID: "s1", At: now, DiskPercent: usedPercent}) d.wg.Wait() } @@ -247,18 +247,18 @@ func TestDispatcherServerUnreachableOncePerOutage(t *testing.T) { d, h := newDispatcher(t) subscribe(t, d, h, "hook", true, ServerUnreachable) ctx := context.Background() - down := store.Server{ID: "s1", Name: "builder-1", Status: store.ServerUnreachable} + down := events.ServerState{ID: "s1", Name: "builder-1", Status: store.ServerUnreachable} // The fleet and the server.state event report the same outage. if err := d.Notify(ctx, ServerUnreachableEvent("", "s1", "builder-1", time.Now())); err != nil { t.Fatal(err) } - d.handle(ctx, events.Event{Type: events.ServerState, Data: down}) + d.serverState(ctx, down) d.wg.Wait() if n := len(h.events()); n != 1 { t.Fatalf("got %d alerts for one outage, want 1", n) } - d.handle(ctx, events.Event{Type: events.ServerState, Data: store.Server{ID: "s1", Name: "builder-1", Status: store.ServerOnline}}) - d.handle(ctx, events.Event{Type: events.ServerState, Data: &down}) + d.serverState(ctx, events.ServerState{ID: "s1", Name: "builder-1", Status: store.ServerOnline}) + d.serverState(ctx, down) d.wg.Wait() if got := h.events(); len(got) != 2 || got[1].Data["server_id"] != "s1" { t.Fatalf("alerts %+v, want a second one for the second outage", got) diff --git a/internal/scheduler/cleanup.go b/internal/scheduler/cleanup.go index d7edb13..6914c42 100644 --- a/internal/scheduler/cleanup.go +++ b/internal/scheduler/cleanup.go @@ -253,7 +253,7 @@ func (s *Scheduler) cancelLostJobs(ctx context.Context) { s.log.Warn("cleanup: complete job", "job", j.ID, "err", err) continue } - s.publish(events.JobState, done) + s.bus.Publish(events.JobState(done)) s.finish(j.ID) } } @@ -285,7 +285,7 @@ func (s *Scheduler) closeUnclaimedJobs(ctx context.Context) { pools := map[string]bool{} for _, j := range jobs { s.log.Info("cleanup: close a job that runplane did not run", "job", j.ID, "queued_at", j.QueuedAt) - s.publish(events.JobState, j) + s.bus.Publish(events.JobState(j)) pools[j.PoolID] = true } for id := range pools { diff --git a/internal/scheduler/cleanup_test.go b/internal/scheduler/cleanup_test.go index 83b6f8f..33ea8f7 100644 --- a/internal/scheduler/cleanup_test.go +++ b/internal/scheduler/cleanup_test.go @@ -249,7 +249,7 @@ func TestCleanupClosesUnclaimedJobs(t *testing.T) { if len(client.acquired) != 0 { t.Fatalf("acquired %v with no room", client.acquired) } - states, stop := e.bus.Subscribe(16) + states := events.Subscribe[events.JobState](t.Context(), e.bus) e.clock.add(59 * time.Minute) e.s.cleanup(e.ctx) @@ -262,10 +262,9 @@ func TestCleanupClosesUnclaimedJobs(t *testing.T) { if j.State != store.JobCompleted || j.Result != store.ResultCancelled || j.RejectReason != store.NotRunReason { t.Errorf("job %s %s %q after 61 minutes, want completed cancelled with %q", j.State, j.Result, j.RejectReason, store.NotRunReason) } - stop() published := false - for ev := range states { - if x, ok := ev.Data.(store.Job); ok && ev.Type == events.JobState && x.ID == j.ID && x.State == store.JobCompleted { + for len(states) > 0 { + if x := <-states; x.ID == j.ID && x.State == store.JobCompleted { published = true } } diff --git a/internal/scheduler/jobs.go b/internal/scheduler/jobs.go index 8ffa1f5..0e0504f 100644 --- a/internal/scheduler/jobs.go +++ b/internal/scheduler/jobs.go @@ -77,7 +77,7 @@ func (s *Scheduler) finishJob(ctx context.Context, jobID string) { s.log.Warn("summarize job metrics", "job", job.ID, "err", err) } if job, err = s.st.GetJob(wctx, job.ID); err == nil { - s.publish(events.JobState, job) + s.bus.Publish(events.JobState(job)) } } @@ -114,7 +114,7 @@ func (s *Scheduler) enrich(job store.Job) { return } if job, err := s.st.GetJob(ctx, job.ID); err == nil { - s.publish(events.JobState, job) + s.bus.Publish(events.JobState(job)) } }) } @@ -165,7 +165,7 @@ func (s *Scheduler) watchSteps(jobID string) { continue } job.Steps = steps - s.publish(events.JobState, job) + s.bus.Publish(events.JobState(job)) } }) } @@ -217,7 +217,7 @@ func (s *Scheduler) hold(ctx context.Context, b store.Sandbox) { return } b.HoldUntil = &until - s.publish(events.SandboxState, b) + s.bus.Publish(events.SandboxState(b)) } // holdForDebug keeps a sandbox that got a debug job after it started without diff --git a/internal/scheduler/resize.go b/internal/scheduler/resize.go index 62faa1b..4e5eda7 100644 --- a/internal/scheduler/resize.go +++ b/internal/scheduler/resize.go @@ -150,7 +150,7 @@ func (s *Scheduler) launch(ctx context.Context, p placement, b store.Sandbox, sp s.log.Warn("insert sandbox", "err", err) return err } - s.publish(events.SandboxState, b) + s.bus.Publish(events.SandboxState(b)) s.publishPoolStats(ctx, b.PoolID) err := p.host.Driver.Start(ctx, spec) diff --git a/internal/scheduler/scale.go b/internal/scheduler/scale.go index 523657e..576a8af 100644 --- a/internal/scheduler/scale.go +++ b/internal/scheduler/scale.go @@ -190,7 +190,7 @@ func (h *scaleSet) handle(ctx context.Context, msg *scaleset.RunnerScaleSetMessa if err != nil { return err } - h.s.publish(events.JobState, job) + h.s.bus.Publish(events.JobState(job)) } for _, m := range msg.JobStartedMessages { if err := h.handleStarted(ctx, m); err != nil { @@ -239,7 +239,7 @@ func (h *scaleSet) handleAvailable(ctx context.Context, list []*scaleset.JobAvai if errors.Is(err, errCheckLater) { // Another job of the message must not wait for this one. s.log.Warn("skip a job until GitHub offers it again", "job", job.ID, "err", err) - s.publish(events.JobState, job) + s.bus.Publish(events.JobState(job)) continue } if err != nil { @@ -260,7 +260,7 @@ func (h *scaleSet) handleAvailable(ctx context.Context, list []*scaleset.JobAvai } job.RejectReason = reason } - s.publish(events.JobState, job) + s.bus.Publish(events.JobState(job)) // A job past queued was acquired before this message came again. if reason == "" && job.State == store.JobQueued { wanted = append(wanted, m) @@ -302,7 +302,7 @@ func (h *scaleSet) handleAvailable(ctx context.Context, list []*scaleset.JobAvai if err != nil { return err } - s.publish(events.JobState, job) + s.bus.Publish(events.JobState(job)) } return nil } @@ -412,7 +412,7 @@ func (h *scaleSet) handleStarted(ctx context.Context, m *scaleset.JobStarted) er s.holdForDebug(ctx, b) } } - s.publish(events.JobState, job) + s.bus.Publish(events.JobState(job)) s.enrich(job) s.watchSteps(job.ID) return nil @@ -444,7 +444,7 @@ func (h *scaleSet) handleCompleted(ctx context.Context, m *scaleset.JobCompleted if job.SandboxID != "" { s.setState(ctx, job.SandboxID, sandbox.StateStopping, "job completed") } - s.publish(events.JobState, job) + s.bus.Publish(events.JobState(job)) // A message that comes again after the log is stored needs no second finish step. if prev.State != store.JobCompleted || prev.LogState != store.LogStored { s.finish(job.ID) diff --git a/internal/scheduler/scale_test.go b/internal/scheduler/scale_test.go index 14acc8c..5b5c7a6 100644 --- a/internal/scheduler/scale_test.go +++ b/internal/scheduler/scale_test.go @@ -360,15 +360,14 @@ func TestPoolStatsEvent(t *testing.T) { e.gh.checks["acme/site"] = JobCheck{Private: false} pool := e.addPool("runplane") _, sc, _ := e.attach(pool) - ch, cancel := e.bus.Subscribe(256) - defer cancel() - last := func() poolStats { + ch := events.Subscribe[events.PoolStats](t.Context(), e.bus) + last := func() events.PoolStats { t.Helper() - var got *poolStats + var got *events.PoolStats for { select { - case ev := <-ch: - if p, ok := ev.Data.(poolStats); ok && ev.Type == events.PoolStats && p.PoolID == pool.ID { + case p := <-ch: + if p.PoolID == pool.ID { got = &p } default: @@ -387,12 +386,12 @@ func TestPoolStatsEvent(t *testing.T) { ) m.JobAvailableMessages[2].QueueTime = start.Add(time.Second) e.scale(sc, m) - if got, want := last(), (poolStats{PoolID: pool.ID, Queued: 1, Assigned: 1}); got != want { + if got, want := last(), (events.PoolStats{PoolID: pool.ID, Queued: 1, Assigned: 1}); got != want { t.Errorf("after the offer: %+v, want %+v", got, want) } e.scale(sc, msg(1)) - if got, want := last(), (poolStats{PoolID: pool.ID, Starting: 1, Queued: 1, Assigned: 1}); got != want { + if got, want := last(), (events.PoolStats{PoolID: pool.ID, Starting: 1, Queued: 1, Assigned: 1}); got != want { t.Errorf("after the start: %+v, want %+v", got, want) } } diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index abdbf1e..5fe460b 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -288,7 +288,7 @@ func (s *Scheduler) setStateAt(ctx context.Context, id string, state sandbox.Sta if !changed { return b, false } - s.publish(events.SandboxState, b) + s.bus.Publish(events.SandboxState(b)) s.publishPoolStats(ctx, b.PoolID) if state.Done() { s.wake(doMaxRunners) @@ -307,27 +307,13 @@ func (s *Scheduler) goBg(fn func(ctx context.Context)) { s.wg.Go(func() { fn(s.bg) }) } -func (s *Scheduler) publish(typ string, data any) { - s.bus.Publish(events.Event{Type: typ, Data: data}) -} - -// poolStats has the JSON fields of the pool.stats event. -type poolStats struct { - PoolID string `json:"pool_id"` - Idle int `json:"idle"` - Busy int `json:"busy"` - Starting int `json:"starting"` - Queued int `json:"queued"` - Assigned int `json:"assigned"` -} - func (s *Scheduler) publishPoolStats(ctx context.Context, poolID string) { counts, err := s.st.CountLiveSandboxes(ctx) if err != nil { s.log.Warn("count sandboxes", "err", err) return } - p := poolStats{PoolID: poolID} + p := events.PoolStats{PoolID: poolID} p.Queued, p.Assigned, err = s.st.CountPoolJobs(ctx, poolID, s.now().Add(-24*time.Hour)) if err != nil { s.log.Warn("count jobs", "err", err) @@ -346,7 +332,7 @@ func (s *Scheduler) publishPoolStats(ctx context.Context, poolID string) { p.Starting += c.Count } } - s.publish(events.PoolStats, p) + s.bus.Publish(p) } // timeLimit is the time limit of a job of the pool. A warm sandbox of the From 46edf3d4d30cdb0f29968f71512c4b221d667ef9 Mon Sep 17 00:00:00 2001 From: ephraimduncan Date: Thu, 5 Nov 2026 20:20:17 +0000 Subject: [PATCH 002/163] refactor(sandbox): one guest contract in internal/sandbox/guest The new package internal/sandbox/guest holds the contract between runplane and runplane-init. Before this change, init, control, the two drivers, and image each had copies of parts of the contract. The package holds: - The guest layout: the ctl folder and socket, the vsock port, the scratch folders, the drive order, the kernel options, the init path, the mount points, the runner user, and the managed label. - The control channel protocol: the paths, the hold body, the shell upgrade names, the status rules, the yamux setup, and the log stream format with a calculated buffer size. - A guest client. init and the tests use it in place of their copies. - The runner patch (Patch and Unpatch) and the cache env. - NewConfig. The two drivers use it in place of their guestConfig and hostname copies. Behavior changes: - The VM driver adds the cache env only for a patched runner, as the container driver does. - The container driver sends only the IPv4 DNS servers in the config. Init uses them only in VM mode. - The config has an ActionsCache field. Init removes the runner patch when the field is false. Before, init looked for ACTIONS_RESULTS_URL. - A nil config replaces the empty JIT config as the sign of a sandbox from before a runplane restart. - The log stream cuts a source name of more than 16 bytes. Before, the host divided a long line with such a name. As a result, sysbox does not import internal/cache, and its build does not include store and SQLite. init does not import control. A new test compares the builtin Dockerfiles with the layout constants. --- cmd/runplane/local.go | 7 +- cmd/runplane/logs.go | 12 +- init/control.go | 138 +++++------------- init/control_test.go | 30 ++-- init/logqueue.go | 38 ++--- init/main_linux.go | 32 ++-- init/overlay_linux_test.go | 10 +- init/pages_linux.go | 4 +- init/requests.go | 14 +- init/runner.go | 36 ----- init/runner_linux.go | 65 ++++----- init/runner_test.go | 40 ----- init/shell.go | 31 +--- init/shell_linux_test.go | 5 +- init/vm_linux.go | 30 ++-- internal/cache/cache.go | 13 +- internal/image/docker.go | 5 +- internal/image/image.go | 3 - internal/image/integration_test.go | 5 +- internal/image/rootdisk.go | 30 +--- internal/image/runner.go | 60 +------- internal/image/runner_integration_test.go | 5 +- internal/image/runner_test.go | 28 ---- internal/sandbox/control/control.go | 130 ++++------------- internal/sandbox/control/logfile.go | 25 +++- internal/sandbox/control/logfile_test.go | 4 +- internal/sandbox/control/tcp.go | 51 +++---- internal/sandbox/control/tcp_test.go | 51 ++----- internal/sandbox/firecracker/api.go | 36 +++-- internal/sandbox/firecracker/firecracker.go | 12 +- .../sandbox/firecracker/firecracker_test.go | 25 ++-- internal/sandbox/firecracker/jail.go | 19 ++- internal/sandbox/firecracker/recover.go | 5 +- internal/sandbox/firecracker/recover_test.go | 21 +-- internal/sandbox/firecracker/start.go | 49 +------ internal/sandbox/firecracker/start_test.go | 87 +++-------- internal/sandbox/firecracker/vm.go | 11 +- internal/sandbox/guest/client.go | 107 ++++++++++++++ internal/sandbox/guest/config.go | 76 ++++++++++ internal/sandbox/guest/config_test.go | 85 +++++++++++ internal/sandbox/guest/dockerfile_test.go | 80 ++++++++++ internal/sandbox/guest/layout.go | 93 ++++++++++++ internal/sandbox/guest/log.go | 98 +++++++++++++ internal/sandbox/guest/log_test.go | 51 +++++++ internal/sandbox/guest/patch.go | 72 +++++++++ internal/sandbox/guest/patch_test.go | 105 +++++++++++++ internal/sandbox/guest/protocol.go | 133 +++++++++++++++++ internal/sandbox/sysbox/driver.go | 6 +- internal/sandbox/sysbox/network.go | 6 +- internal/sandbox/sysbox/probe.go | 5 +- internal/sandbox/sysbox/recover.go | 3 +- internal/sandbox/sysbox/spec.go | 49 +------ internal/sandbox/sysbox/spec_test.go | 60 -------- internal/sandbox/sysbox/start.go | 20 ++- 54 files changed, 1287 insertions(+), 929 deletions(-) create mode 100644 internal/sandbox/guest/client.go create mode 100644 internal/sandbox/guest/config.go create mode 100644 internal/sandbox/guest/config_test.go create mode 100644 internal/sandbox/guest/dockerfile_test.go create mode 100644 internal/sandbox/guest/layout.go create mode 100644 internal/sandbox/guest/log.go create mode 100644 internal/sandbox/guest/log_test.go create mode 100644 internal/sandbox/guest/patch.go create mode 100644 internal/sandbox/guest/patch_test.go create mode 100644 internal/sandbox/guest/protocol.go diff --git a/cmd/runplane/local.go b/cmd/runplane/local.go index a10a758..d8c76e4 100644 --- a/cmd/runplane/local.go +++ b/cmd/runplane/local.go @@ -15,7 +15,6 @@ import ( "github.com/runplane/runplane/images" "github.com/runplane/runplane/internal/agent" - "github.com/runplane/runplane/internal/cache" "github.com/runplane/runplane/internal/config" "github.com/runplane/runplane/internal/host" "github.com/runplane/runplane/internal/image" @@ -101,11 +100,11 @@ func newLocal(ctx context.Context, cfg config.Config, lc localConfig, log *slog. case sandbox.ModeVM: drv, err := firecracker.New(firecracker.Options{ DataDir: cfg.DataDir, Kernel: cfg.KernelPath(), ServicePort: cfg.ServicePort, RootDisk: images.RootDisk, - RunnerDrive: func(ctx context.Context, version, arch string) (string, error) { + RunnerDrive: func(ctx context.Context, version, arch string) (string, bool, error) { b, err := bundles.Ensure(ctx, version, arch) - return b.Ext4Path, err + return b.Ext4Path, b.Patched, err }, - CacheEnv: cache.RunnerEnv, Settings: lc.settings, Logger: log.With("driver", "firecracker"), + Settings: lc.settings, Logger: log.With("driver", "firecracker"), }) if err != nil { return nil, fmt.Errorf("start the VM driver: %w", err) diff --git a/cmd/runplane/logs.go b/cmd/runplane/logs.go index d105ae7..351a924 100644 --- a/cmd/runplane/logs.go +++ b/cmd/runplane/logs.go @@ -14,6 +14,8 @@ import ( "github.com/runplane/runplane/internal/config" "github.com/runplane/runplane/internal/sandbox" + "github.com/runplane/runplane/internal/sandbox/control" + "github.com/runplane/runplane/internal/sandbox/guest" "github.com/runplane/runplane/internal/store" ) @@ -84,14 +86,10 @@ func (l logs) sandboxLog(ctx context.Context, b store.Sandbox, follow bool) (io. return drv.Logs(ctx, b.ID, follow) } -// diagLines keeps the step output of a sandbox log line -// "