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()