From bdd0f09e5e42e2a74822c021a50664a13b5ae0ed Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 2 Oct 2026 14:47:37 +0800 Subject: [PATCH] fix(daemon): /v1/vms and /v1/events wait for the first reconcile pass The API socket opens before the first pass publishes, so a client that connected in that window got an empty VM list and an empty opening sync, then an ADDED change for every VM. Both handlers now wait until the first pass publishes, healthy or not, or the client leaves. handleEvents waits before subscribing, so the first pass does not replay as ADDED after the sync; the existing subscribe-then-snapshot order keeps later changes from being missed. healthz is unchanged and still reports unready until then. --- daemon/api.go | 8 +++- daemon/api_test.go | 92 ++++++++++++++++++++++++++++++++++++++++++++++ daemon/cache.go | 18 ++++++++- 3 files changed, 116 insertions(+), 2 deletions(-) create mode 100644 daemon/api_test.go diff --git a/daemon/api.go b/daemon/api.go index e1def776..28675dcb 100644 --- a/daemon/api.go +++ b/daemon/api.go @@ -102,13 +102,19 @@ func (d *Daemon) handleHealth(w http.ResponseWriter, _ *http.Request) { writeJSON(w, status, healthResponse{OK: ok, Degraded: h.degraded, LastPass: h.lastPass, VMs: len(all)}) } -func (d *Daemon) handleVMs(w http.ResponseWriter, _ *http.Request) { +func (d *Daemon) handleVMs(w http.ResponseWriter, r *http.Request) { + if d.state.waitReady(r.Context()) != nil { + return + } all, _ := d.state.snapshot() writeJSON(w, http.StatusOK, vmsResponse{VMs: all}) } // handleEvents streams changes after an opening snapshot; it replays nothing, so a reconnecting client diffs per-VM generations against the fresh snapshot. func (d *Daemon) handleEvents(w http.ResponseWriter, r *http.Request) { + if d.state.waitReady(r.Context()) != nil { + return + } events, release := d.state.subscribe() defer release() diff --git a/daemon/api_test.go b/daemon/api_test.go new file mode 100644 index 00000000..25f59fa2 --- /dev/null +++ b/daemon/api_test.go @@ -0,0 +1,92 @@ +package daemon + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + "testing/synctest" + "time" + + "github.com/cocoonstack/cocoon/types" +) + +func TestReadsWaitForTheFirstPass(t *testing.T) { + tests := []struct { + name string + path string + want string + }{ + {"vms", "/v1/vms", `{"vms":[{"backend":"fake-hv","id":"vm1"`}, + {"events", "/v1/events", "event: sync\ndata: {\"vms\":[{\"backend\":\"fake-hv\",\"id\":\"vm1\""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + d := &Daemon{state: newCache()} + ctx, cancel := context.WithCancel(t.Context()) + rec, done := serve(ctx, d, tt.path) + + synctest.Wait() + if rec.Body.Len() != 0 { + t.Fatalf("answered before the first pass: %q", rec.Body.String()) + } + + st := statusOf("vm1", types.VMStateRunning, 1, true) + st.Backend = "fake-hv" + d.state.publish([]VMStatus{st}, true, 0, time.Now()) + synctest.Wait() + body := rec.Body.String() + if !strings.HasPrefix(body, tt.want) { + t.Fatalf("got %q, want prefix %q", body, tt.want) + } + if strings.Contains(body, "event: change") { + t.Fatalf("the first pass replayed as changes after the sync: %q", body) + } + + cancel() + <-done + }) + }) + } +} + +func TestReadsUnblockOnAFailedFirstPass(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + d := &Daemon{state: newCache()} + rec, done := serve(t.Context(), d, "/v1/vms") + + synctest.Wait() + d.state.publish(nil, false, 0, time.Now()) + <-done + if got := rec.Body.String(); got != "{\"vms\":[]}\n" { + t.Fatalf("got %q, want the empty list the failed pass published", got) + } + }) +} + +func TestReadsReturnWhenTheClientLeavesBeforeTheFirstPass(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + d := &Daemon{state: newCache()} + ctx, cancel := context.WithCancel(t.Context()) + rec, done := serve(ctx, d, "/v1/events") + + synctest.Wait() + cancel() + <-done + if rec.Body.Len() != 0 { + t.Fatalf("wrote to a client that left before the first pass: %q", rec.Body.String()) + } + }) +} + +func serve(ctx context.Context, d *Daemon, path string) (*httptest.ResponseRecorder, <-chan struct{}) { + rec := httptest.NewRecorder() + done := make(chan struct{}) + go func() { + defer close(done) + d.routes().ServeHTTP(rec, httptest.NewRequestWithContext(ctx, http.MethodGet, path, nil)) + }() + return rec, done +} diff --git a/daemon/cache.go b/daemon/cache.go index 6fcbfaa4..2ff7392e 100644 --- a/daemon/cache.go +++ b/daemon/cache.go @@ -1,6 +1,7 @@ package daemon import ( + "context" "maps" "slices" "sync" @@ -81,13 +82,14 @@ type cache struct { healthy bool degraded int lastPass time.Time + ready chan struct{} nextSub int subs map[int]chan changeEvent } func newCache() *cache { - return &cache{byKey: map[watchKey]VMStatus{}, subs: map[int]chan changeEvent{}} + return &cache{byKey: map[watchKey]VMStatus{}, ready: make(chan struct{}), subs: map[int]chan changeEvent{}} } // publish replaces the snapshot and emits the diff against the previous pass. @@ -115,6 +117,11 @@ func (c *cache) publish(all []VMStatus, healthy bool, degraded int, at time.Time } c.byKey, c.order = next, all c.healthy, c.degraded, c.lastPass = healthy, degraded, at + select { + case <-c.ready: + default: + close(c.ready) + } subs := slices.Collect(maps.Values(c.subs)) c.mu.Unlock() @@ -140,6 +147,15 @@ func (c *cache) byBackend(backend string) []VMStatus { return out } +func (c *cache) waitReady(ctx context.Context) error { + select { + case <-c.ready: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + func (c *cache) snapshot() ([]VMStatus, health) { c.mu.RLock() defer c.mu.RUnlock()