fix(daemon): /v1/vms and /v1/events wait for the first reconcile pass

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

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
+}

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