From 44e2cb1303910d99f25166956030d4cdaa0fa592 Mon Sep 17 00:00:00 2001 From: Jon Chery Date: Sat, 1 Aug 2026 19:56:39 +0000 Subject: [PATCH] ship(P01): iter.Seq streaming merged into v0.3 milestone ---ci--- project: orca phase: 1 milestone: v0.3 status: complete requirements: covered: [REQ-022, REQ-030] partial: [] ---/ci--- P01: iter.Seq streaming for --watch flags. - JobRepo.Watch / NodeRepo.Watch: pull-based iter.Seq[[]*T] snapshot-per-tick (G-001) - Immediate first yield before ticker (G-002) - Table mode: clear-screen + re-render on change - JSON mode: init/update/delete events, one line per change - signal.NotifyContext on SIGINT/SIGTERM (D-023) - 16 tests (8 store + 8 CLI), all pass under -race - 4-layer verification passed --- .ciagent/CHECKPOINT.json | 8 +- internal/cli/job.go | 77 ++++++- internal/cli/node.go | 77 ++++++- internal/cli/watch_test.go | 287 +++++++++++++++++++++++++++ internal/store/job_task_repo.go | 54 +++++ internal/store/job_task_repo_test.go | 201 +++++++++++++++++++ internal/store/node_repo.go | 36 ++++ internal/store/node_repo_test.go | 156 +++++++++++++++ 8 files changed, 890 insertions(+), 6 deletions(-) create mode 100644 internal/cli/watch_test.go create mode 100644 internal/store/job_task_repo_test.go diff --git a/.ciagent/CHECKPOINT.json b/.ciagent/CHECKPOINT.json index 1748a51..e6ec172 100644 --- a/.ciagent/CHECKPOINT.json +++ b/.ciagent/CHECKPOINT.json @@ -1,9 +1,9 @@ { - "phase": 0, - "stage": "grill", + "phase": 1, + "stage": "verify", "milestone": "v0.3", "milestone_slug": "scheduling-streaming", - "phase_role": "pre_execution", + "phase_role": "execution", "attempts": 0, - "updated_at": "2026-08-01T00:04:00Z" + "updated_at": "2026-08-01T00:10:00Z" } \ No newline at end of file diff --git a/internal/cli/job.go b/internal/cli/job.go index bfced37..86ce40d 100644 --- a/internal/cli/job.go +++ b/internal/cli/job.go @@ -5,6 +5,9 @@ import ( "encoding/json" "errors" "fmt" + "os" + "os/signal" + "syscall" "time" "github.com/google/uuid" @@ -36,6 +39,7 @@ var ( stopID string runTarget string runIDKey string + jobWatch bool ) var jobRunCmd = &cobra.Command{ @@ -112,8 +116,11 @@ var jobRunCmd = &cobra.Command{ var jobListCmd = &cobra.Command{ Use: "list", Short: "List all jobs", - Long: "Display all jobs and their status.", + Long: "Display all jobs and their status. Use --watch to stream updates until Ctrl-C.", RunE: func(cmd *cobra.Command, args []string) error { + if jobWatch { + return watchJobs(cmd) + } ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second) defer cancel() @@ -142,6 +149,73 @@ var jobListCmd = &cobra.Command{ }, } +func watchJobs(cmd *cobra.Command) error { + ctx, cancel := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM) + defer cancel() + return watchJobsCtx(cmd, ctx) +} + +func watchJobsCtx(cmd *cobra.Command, ctx context.Context) error { + db, closer, err := openDB() + if err != nil { + return err + } + defer closer() + + out := cmd.OutOrStdout() + + if jsonOutput { + seen := make(map[string]string) + for snapshot := range store.NewJobRepo(db).Watch(ctx) { + current := make(map[string]bool, len(snapshot)) + for _, j := range snapshot { + current[j.ID] = true + compact, _ := json.Marshal(j) + key := string(compact) + if prev, ok := seen[j.ID]; !ok || prev != key { + event := "init" + if ok { + event = "update" + } + line, _ := json.Marshal(map[string]any{"event": event, "job": j}) + fmt.Fprintln(out, string(line)) + seen[j.ID] = key + } + } + for id := range seen { + if !current[id] { + line, _ := json.Marshal(map[string]any{"event": "delete", "id": id}) + fmt.Fprintln(out, string(line)) + delete(seen, id) + } + } + } + return nil + } + + prevTable := "" + for snapshot := range store.NewJobRepo(db).Watch(ctx) { + table := renderJobTable(snapshot) + if table != prevTable { + fmt.Fprint(out, "\033[2J\033[H") + fmt.Fprint(out, table) + prevTable = table + } + } + return nil +} + +func renderJobTable(jobs []*model.Job) string { + if len(jobs) == 0 { + return "No jobs.\n" + } + out := fmt.Sprintf("%-36s %-20s %-12s %-8s\n", "ID", "NAME", "STATUS", "EXIT") + for _, j := range jobs { + out += fmt.Sprintf("%-36s %-20s %-12s %-8d\n", j.ID, j.Name, j.Status, j.ExitCode) + } + return out +} + var jobStopCmd = &cobra.Command{ Use: "stop [job-id]", Short: "Stop a running job", @@ -235,6 +309,7 @@ func init() { jobLogsCmd.Flags().StringVar(&stopID, "id", "", "job id") jobRunCmd.Flags().StringVar(&runTarget, "target", "", "pin job to a specific node id (overrides bin-packing)") jobRunCmd.Flags().StringVar(&runIDKey, "idempotency-key", "", "X-Orca-Idempotency-Key for cross-node dispatch dedupe") + jobListCmd.Flags().BoolVar(&jobWatch, "watch", false, "stream jobs until Ctrl-C (table refresh or --json per-event)") jobCmd.AddCommand(jobRunCmd) jobCmd.AddCommand(jobListCmd) diff --git a/internal/cli/node.go b/internal/cli/node.go index c867c0b..9b9123a 100644 --- a/internal/cli/node.go +++ b/internal/cli/node.go @@ -3,10 +3,13 @@ package cli import ( "context" "database/sql" + "encoding/json" "fmt" "log/slog" "os" + "os/signal" "path/filepath" + "syscall" "time" "github.com/google/uuid" @@ -54,6 +57,7 @@ var ( joinAddr string joinCAFinger string leaveID string + nodeWatch bool ) var nodeCmd = &cobra.Command{ @@ -155,8 +159,11 @@ var nodeLeaveCmd = &cobra.Command{ var nodeListCmd = &cobra.Command{ Use: "list", Short: "List all nodes in the orca registry", - Long: "Display all registered nodes and their state.", + Long: "Display all registered nodes and their state. Use --watch to stream updates until Ctrl-C.", RunE: func(cmd *cobra.Command, args []string) error { + if nodeWatch { + return watchNodes(cmd) + } ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second) defer cancel() @@ -185,11 +192,79 @@ var nodeListCmd = &cobra.Command{ }, } +func watchNodes(cmd *cobra.Command) error { + ctx, cancel := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM) + defer cancel() + return watchNodesCtx(cmd, ctx) +} + +func watchNodesCtx(cmd *cobra.Command, ctx context.Context) error { + db, closer, err := openDB() + if err != nil { + return err + } + defer closer() + + out := cmd.OutOrStdout() + + if jsonOutput { + seen := make(map[string]string) + for snapshot := range store.NewNodeRepo(db).Watch(ctx) { + current := make(map[string]bool, len(snapshot)) + for _, n := range snapshot { + current[n.ID] = true + compact, _ := json.Marshal(n) + key := string(compact) + if prev, ok := seen[n.ID]; !ok || prev != key { + event := "init" + if ok { + event = "update" + } + line, _ := json.Marshal(map[string]any{"event": event, "node": n}) + fmt.Fprintln(out, string(line)) + seen[n.ID] = key + } + } + for id := range seen { + if !current[id] { + line, _ := json.Marshal(map[string]any{"event": "delete", "id": id}) + fmt.Fprintln(out, string(line)) + delete(seen, id) + } + } + } + return nil + } + + prevTable := "" + for snapshot := range store.NewNodeRepo(db).Watch(ctx) { + table := renderNodeTable(snapshot) + if table != prevTable { + fmt.Fprint(out, "\033[2J\033[H") + fmt.Fprint(out, table) + prevTable = table + } + } + return nil +} + +func renderNodeTable(nodes []*model.Node) string { + if len(nodes) == 0 { + return "No nodes registered.\n" + } + out := fmt.Sprintf("%-36s %-20s %-22s %-10s\n", "ID", "NAME", "ADDRESS", "STATE") + for _, n := range nodes { + out += fmt.Sprintf("%-36s %-20s %-22s %-10s\n", n.ID, n.Name, n.Address, n.State) + } + return out +} + func init() { nodeJoinCmd.Flags().StringVar(&joinName, "name", "", "node name (required)") nodeJoinCmd.Flags().StringVar(&joinAddr, "addr", "", "node address (default localhost:8443)") nodeJoinCmd.Flags().StringVar(&joinCAFinger, "ca-fingerprint", "", "pin CA cert SHA-256 (REQ-026); fails if on-disk CA doesn't match") nodeLeaveCmd.Flags().StringVar(&leaveID, "id", "", "node id") + nodeListCmd.Flags().BoolVar(&nodeWatch, "watch", false, "stream nodes until Ctrl-C (table refresh or --json per-event)") nodeCmd.AddCommand(nodeJoinCmd) nodeCmd.AddCommand(nodeLeaveCmd) diff --git a/internal/cli/watch_test.go b/internal/cli/watch_test.go new file mode 100644 index 0000000..a1fdadd --- /dev/null +++ b/internal/cli/watch_test.go @@ -0,0 +1,287 @@ +package cli + +import ( + "bytes" + "context" + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "git.cloudinit.dev/coreci/orca/internal/model" + "git.cloudinit.dev/coreci/orca/internal/store" +) + +func TestWatchJobs_JSONStreaming(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "orca.db") + db, err := store.Open(dbPath) + if err != nil { + t.Fatalf("open db: %v", err) + } + defer db.Close() + + repo := store.NewJobRepo(db) + bgCtx := context.Background() + _ = repo.Insert(bgCtx, &model.Job{ID: "seed-job", Name: "seed", Spec: "t", Status: model.JobStatusPending}) + + t.Setenv("ORCA_DB", dbPath) + + jsonOutput = true + t.Cleanup(func() { jsonOutput = false }) + + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + t.Cleanup(func() { rootCmd.SetOut(os.Stdout); rootCmd.SetErr(os.Stderr) }) + + ctx, cancel := context.WithCancel(bgCtx) + + done := make(chan error, 1) + go func() { done <- watchJobsCtx(rootCmd, ctx) }() + + // First yield is immediate (G-002); wait for it. + time.Sleep(100 * time.Millisecond) + + _ = repo.Insert(bgCtx, &model.Job{ID: "watch-job", Name: "watch", Spec: "t", Status: model.JobStatusPending}) + + // Wait for at least one ticker interval (default 1s) to capture the change. + time.Sleep(1100 * time.Millisecond) + cancel() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("watchJobsCtx did not return within 2s after cancel") + } + + output := buf.String() + if !strings.Contains(output, `"event":"init"`) { + t.Errorf("expected init event, got: %s", output) + } + if !strings.Contains(output, "watch-job") { + t.Errorf("expected watch-job in output, got: %s", output) + } +} + +func TestWatchJobs_TableRefresh(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "orca.db") + db, err := store.Open(dbPath) + if err != nil { + t.Fatalf("open db: %v", err) + } + defer db.Close() + + repo := store.NewJobRepo(db) + bgCtx := context.Background() + _ = repo.Insert(bgCtx, &model.Job{ID: "seed-job", Name: "seed", Spec: "t", Status: model.JobStatusPending}) + + t.Setenv("ORCA_DB", dbPath) + + jsonOutput = false + t.Cleanup(func() { jsonOutput = false }) + + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + t.Cleanup(func() { rootCmd.SetOut(os.Stdout); rootCmd.SetErr(os.Stderr) }) + + ctx, cancel := context.WithCancel(bgCtx) + + done := make(chan error, 1) + go func() { done <- watchJobsCtx(rootCmd, ctx) }() + + time.Sleep(100 * time.Millisecond) + + _ = repo.Insert(bgCtx, &model.Job{ID: "table-job", Name: "table", Spec: "t", Status: model.JobStatusPending}) + + time.Sleep(1100 * time.Millisecond) + cancel() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("watchJobsCtx did not return within 2s after cancel") + } + + output := buf.String() + if !strings.Contains(output, "\033[2J\033[H") { + t.Errorf("expected clear-screen escape in table watch output, got: %s", output) + } + if !strings.Contains(output, "table-job") { + t.Errorf("expected table-job in output, got: %s", output) + } +} + +func TestWatchNodes_JSONStreaming(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "orca.db") + db, err := store.Open(dbPath) + if err != nil { + t.Fatalf("open db: %v", err) + } + defer db.Close() + + repo := store.NewNodeRepo(db) + bgCtx := context.Background() + _ = repo.Insert(bgCtx, &model.Node{ + ID: "seed-node", Name: "seed", Address: "addr", + State: model.NodeStateReady, JoinedAt: time.Now().UTC(), LastSeen: time.Now().UTC(), + }) + + t.Setenv("ORCA_DB", dbPath) + + jsonOutput = true + t.Cleanup(func() { jsonOutput = false }) + + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + t.Cleanup(func() { rootCmd.SetOut(os.Stdout); rootCmd.SetErr(os.Stderr) }) + + ctx, cancel := context.WithCancel(bgCtx) + + done := make(chan error, 1) + go func() { done <- watchNodesCtx(rootCmd, ctx) }() + + time.Sleep(100 * time.Millisecond) + + _ = repo.Insert(bgCtx, &model.Node{ + ID: "watch-node", Name: "watch", Address: "addr2", + State: model.NodeStateReady, JoinedAt: time.Now().UTC(), LastSeen: time.Now().UTC(), + }) + + time.Sleep(1100 * time.Millisecond) + cancel() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("watchNodesCtx did not return within 2s after cancel") + } + + output := buf.String() + initFound := false + watchNodeFound := false + for _, line := range strings.Split(output, "\n") { + line = strings.TrimSpace(line) + if line == "" { + continue + } + var event map[string]any + if err := json.Unmarshal([]byte(line), &event); err != nil { + continue + } + if event["event"] == "init" { + initFound = true + if node, ok := event["node"].(map[string]any); ok { + if node["id"] == "watch-node" { + watchNodeFound = true + } + } + } + } + if !initFound { + t.Errorf("expected init event in JSON stream, got: %s", output) + } + if !watchNodeFound { + t.Errorf("expected watch-node in JSON stream, got: %s", output) + } +} + +func TestWatchNodes_TableRefresh(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "orca.db") + db, err := store.Open(dbPath) + if err != nil { + t.Fatalf("open db: %v", err) + } + defer db.Close() + + repo := store.NewNodeRepo(db) + bgCtx := context.Background() + _ = repo.Insert(bgCtx, &model.Node{ + ID: "seed-node", Name: "seed", Address: "addr", + State: model.NodeStateReady, JoinedAt: time.Now().UTC(), LastSeen: time.Now().UTC(), + }) + + t.Setenv("ORCA_DB", dbPath) + + jsonOutput = false + t.Cleanup(func() { jsonOutput = false }) + + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + t.Cleanup(func() { rootCmd.SetOut(os.Stdout); rootCmd.SetErr(os.Stderr) }) + + ctx, cancel := context.WithCancel(bgCtx) + + done := make(chan error, 1) + go func() { done <- watchNodesCtx(rootCmd, ctx) }() + + time.Sleep(100 * time.Millisecond) + + _ = repo.Insert(bgCtx, &model.Node{ + ID: "table-node", Name: "table", Address: "addr2", + State: model.NodeStateReady, JoinedAt: time.Now().UTC(), LastSeen: time.Now().UTC(), + }) + + time.Sleep(1100 * time.Millisecond) + cancel() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("watchNodesCtx did not return within 2s after cancel") + } + + output := buf.String() + if !strings.Contains(output, "\033[2J\033[H") { + t.Errorf("expected clear-screen escape in table watch output, got: %s", output) + } + if !strings.Contains(output, "table-node") { + t.Errorf("expected table-node in output, got: %s", output) + } +} + +func TestRenderJobTable(t *testing.T) { + jobs := []*model.Job{ + {ID: "j1", Name: "alpha", Status: "running", ExitCode: 0}, + {ID: "j2", Name: "beta", Status: "done", ExitCode: 0}, + } + out := renderJobTable(jobs) + if !strings.Contains(out, "j1") || !strings.Contains(out, "alpha") { + t.Errorf("renderJobTable missing job 1: %s", out) + } + if !strings.Contains(out, "j2") || !strings.Contains(out, "beta") { + t.Errorf("renderJobTable missing job 2: %s", out) + } +} + +func TestRenderJobTableEmpty(t *testing.T) { + out := renderJobTable(nil) + if !strings.Contains(out, "No jobs") { + t.Errorf("expected empty message, got: %s", out) + } +} + +func TestRenderNodeTable(t *testing.T) { + nodes := []*model.Node{ + {ID: "n1", Name: "alpha", Address: "localhost:8443", State: "ready"}, + } + out := renderNodeTable(nodes) + if !strings.Contains(out, "n1") || !strings.Contains(out, "alpha") { + t.Errorf("renderNodeTable missing node: %s", out) + } +} + +func TestRenderNodeTableEmpty(t *testing.T) { + out := renderNodeTable(nil) + if !strings.Contains(out, "No nodes") { + t.Errorf("expected empty message, got: %s", out) + } +} diff --git a/internal/store/job_task_repo.go b/internal/store/job_task_repo.go index b509f8b..4170e12 100644 --- a/internal/store/job_task_repo.go +++ b/internal/store/job_task_repo.go @@ -6,11 +6,19 @@ import ( "encoding/json" "errors" "fmt" + "iter" + "log/slog" "time" "git.cloudinit.dev/coreci/orca/internal/model" ) +// watchInterval is the poll cadence used by JobRepo.Watch and NodeRepo.Watch. +// It is an unexported package var (default 1s) so tests can override it to a +// small value for deterministic assertions (D-043). Do not change it from +// production code paths. +var watchInterval = 1 * time.Second + type JobRepo struct { db *sql.DB } @@ -59,6 +67,52 @@ func (r *JobRepo) List(ctx context.Context) ([]*model.Job, error) { return jobs, rows.Err() } +// Watch yields the full snapshot of jobs on a watchInterval ticker until ctx +// is cancelled or the consumer stops pulling (yield returns false). It does +// not spawn a goroutine; the polling loop runs inline in the caller's +// goroutine via the range-over-func pull protocol (D-032). +// +// Each tick re-runs the List query and yields one []*model.Job snapshot +// containing ALL rows for that tick (G-001). The first yield happens +// immediately before the first ticker wait, so the consumer sees the initial +// state with no watchInterval delay (G-002). Transient query/scan errors are +// logged via slog.Default().Warn and the loop continues to the next tick +// rather than terminating the stream (D-034 lite). The ticker is stopped and +// rows are closed on every exit path (ctx.Done, yield==false, scan error). +func (r *JobRepo) Watch(ctx context.Context) iter.Seq[[]*model.Job] { + return func(yield func([]*model.Job) bool) { + ticker := time.NewTicker(watchInterval) + defer ticker.Stop() + for { + rows, err := r.db.QueryContext(ctx, + `SELECT id, name, spec, status, exit_code, created_at, started_at, ended_at FROM jobs ORDER BY created_at DESC`) + if err != nil { + slog.Default().Warn("watch jobs: query failed", "error", err) + // fall through to the select to wait for the next tick + } else { + snapshot := make([]*model.Job, 0) + for rows.Next() { + j, scanErr := scanJob(rows) + if scanErr != nil { + slog.Default().Warn("watch jobs: scan failed", "error", scanErr) + continue + } + snapshot = append(snapshot, j) + } + rows.Close() + if !yield(snapshot) { + return // consumer stopped pulling + } + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + } +} + func (r *JobRepo) UpdateStatus(ctx context.Context, id string, status model.JobStatus, exitCode int) error { now := time.Now().UTC() var startedAt, endedAt *time.Time diff --git a/internal/store/job_task_repo_test.go b/internal/store/job_task_repo_test.go new file mode 100644 index 0000000..a126cfd --- /dev/null +++ b/internal/store/job_task_repo_test.go @@ -0,0 +1,201 @@ +package store + +import ( + "context" + "path/filepath" + "testing" + "time" + + "git.cloudinit.dev/coreci/orca/internal/model" +) + +func openJobTestDB(t *testing.T) (*JobRepo, func()) { + t.Helper() + path := filepath.Join(t.TempDir(), "test.db") + db, err := Open(path) + if err != nil { + t.Fatalf("open db: %v", err) + } + return NewJobRepo(db), func() { _ = db.Close() } +} + +func insertJob(t *testing.T, repo *JobRepo, ctx context.Context, id, name string) { + t.Helper() + if err := repo.Insert(ctx, &model.Job{ + ID: id, + Name: name, + Spec: "test", + Status: model.JobStatusPending, + }); err != nil { + t.Fatalf("insert job %s: %v", id, err) + } +} + +// withFastWatch sets watchInterval to a small value for deterministic tests and +// restores the default (1s) on cleanup. +func withFastWatch(t *testing.T, d time.Duration) { + t.Helper() + prev := watchInterval + watchInterval = d + t.Cleanup(func() { watchInterval = prev }) +} + +// TestJobRepoWatch_YieldsSnapshots verifies each yield is a complete tick +// snapshot (G-001): the first snapshot contains only the first job, and a +// later snapshot contains both jobs after a second insert. +func TestJobRepoWatch_YieldsSnapshots(t *testing.T) { + withFastWatch(t, 10*time.Millisecond) + repo, cleanup := openJobTestDB(t) + defer cleanup() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + insertJob(t, repo, ctx, "job-1", "alpha") + + var snapshots [][]*model.Job + done := make(chan struct{}) + go func() { + defer close(done) + for snap := range repo.Watch(ctx) { + snapshots = append(snapshots, snap) + if len(snapshots) >= 15 { + cancel() + return + } + } + }() + + // Insert a second job after a short delay so a later tick observes it. + // Use a background context — the watch ctx may be cancelled by the + // goroutine above once it collects enough snapshots. + time.Sleep(100 * time.Millisecond) + insertJob(t, repo, context.Background(), "job-2", "beta") + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("watch did not complete within 2s") + } + + if len(snapshots) == 0 { + t.Fatal("expected at least one snapshot, got none") + } + // First snapshot must contain only the first job (G-001). + if len(snapshots[0]) != 1 || snapshots[0][0].ID != "job-1" { + t.Errorf("first snapshot = %+v, want only job-1", snapshots[0]) + } + // At least one later snapshot must contain both jobs. + foundBoth := false + for _, snap := range snapshots[1:] { + ids := make(map[string]bool, len(snap)) + for _, j := range snap { + ids[j.ID] = true + } + if ids["job-1"] && ids["job-2"] { + foundBoth = true + break + } + } + if !foundBoth { + t.Errorf("no snapshot contained both jobs; snapshots=%v", snapshots) + } +} + +// TestJobRepoWatch_ImmediateFirstYield verifies G-002: the first snapshot +// arrives before the first ticker wait, i.e. well under the watchInterval. +func TestJobRepoWatch_ImmediateFirstYield(t *testing.T) { + withFastWatch(t, 200*time.Millisecond) + repo, cleanup := openJobTestDB(t) + defer cleanup() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + insertJob(t, repo, ctx, "job-immediate", "first") + + start := time.Now() + var firstSnap []*model.Job + got := make(chan struct{}) + go func() { + for snap := range repo.Watch(ctx) { + firstSnap = snap + close(got) + cancel() + return + } + }() + + select { + case <-got: + case <-time.After(100 * time.Millisecond): + t.Fatal("first yield took >100ms; expected immediate (G-002)") + } + + elapsed := time.Since(start) + if elapsed > 100*time.Millisecond { + t.Errorf("first yield took %v; expected immediate (G-002)", elapsed) + } + if len(firstSnap) != 1 || firstSnap[0].ID != "job-immediate" { + t.Errorf("first snapshot = %+v, want job-immediate", firstSnap) + } +} + +// TestJobRepoWatch_StopsOnConsumerBreak verifies the yield==false path: the +// range loop returns promptly when the consumer breaks after the first yield. +func TestJobRepoWatch_StopsOnConsumerBreak(t *testing.T) { + withFastWatch(t, 10*time.Millisecond) + repo, cleanup := openJobTestDB(t) + defer cleanup() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + insertJob(t, repo, ctx, "job-break", "break") + + done := make(chan struct{}) + go func() { + defer close(done) + for range repo.Watch(ctx) { + break // stop pulling immediately after the first snapshot + } + }() + + select { + case <-done: + // success: range returned + case <-time.After(500 * time.Millisecond): + t.Fatal("watch did not stop on consumer break within 500ms") + } +} + +// TestJobRepoWatch_StopsOnCtxCancel verifies the loop exits promptly after +// ctx is cancelled. +func TestJobRepoWatch_StopsOnCtxCancel(t *testing.T) { + withFastWatch(t, 10*time.Millisecond) + repo, cleanup := openJobTestDB(t) + defer cleanup() + + ctx, cancel := context.WithCancel(context.Background()) + + insertJob(t, repo, ctx, "job-cancel", "cancel") + + done := make(chan struct{}) + go func() { + defer close(done) + for range repo.Watch(ctx) { + // drain until cancelled + } + }() + + // Let at least one tick land, then cancel. + time.Sleep(20 * time.Millisecond) + cancel() + + select { + case <-done: + // success + case <-time.After(500 * time.Millisecond): + t.Fatal("watch did not stop on ctx cancel within 500ms") + } +} diff --git a/internal/store/node_repo.go b/internal/store/node_repo.go index 82cc4d1..70343a8 100644 --- a/internal/store/node_repo.go +++ b/internal/store/node_repo.go @@ -6,6 +6,8 @@ import ( "encoding/json" "errors" "fmt" + "iter" + "log/slog" "time" "git.cloudinit.dev/coreci/orca/internal/model" @@ -69,6 +71,40 @@ func (r *NodeRepo) List(ctx context.Context) ([]*model.Node, error) { return nodes, rows.Err() } +func (r *NodeRepo) Watch(ctx context.Context) iter.Seq[[]*model.Node] { + return func(yield func([]*model.Node) bool) { + ticker := time.NewTicker(watchInterval) + defer ticker.Stop() + for { + rows, err := r.db.QueryContext(ctx, + `SELECT id, name, address, state, joined_at, last_seen, metadata FROM nodes ORDER BY joined_at ASC`) + if err != nil { + slog.Default().Warn("watch nodes: query failed", "error", err) + // fall through to the select to wait for the next tick + } else { + snapshot := make([]*model.Node, 0) + for rows.Next() { + n, scanErr := scanNode(rows) + if scanErr != nil { + slog.Default().Warn("watch nodes: scan failed", "error", scanErr) + continue + } + snapshot = append(snapshot, n) + } + rows.Close() + if !yield(snapshot) { + return // consumer stopped pulling + } + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + } +} + func (r *NodeRepo) UpdateState(ctx context.Context, id string, state model.NodeState) error { res, err := r.db.ExecContext(ctx, `UPDATE nodes SET state = ?, last_seen = ? WHERE id = ?`, diff --git a/internal/store/node_repo_test.go b/internal/store/node_repo_test.go index 711f10a..a7d8a5a 100644 --- a/internal/store/node_repo_test.go +++ b/internal/store/node_repo_test.go @@ -100,3 +100,159 @@ func TestNodeRepo_Delete(t *testing.T) { t.Errorf("expected ErrNotFound, got %v", err) } } + +func insertNode(t *testing.T, repo *NodeRepo, ctx context.Context, id, name string) { + t.Helper() + if err := repo.Insert(ctx, &model.Node{ + ID: id, + Name: name, + Address: "addr", + State: model.NodeStateReady, + JoinedAt: time.Now().UTC(), + LastSeen: time.Now().UTC(), + }); err != nil { + t.Fatalf("insert node %s: %v", id, err) + } +} + +func TestNodeRepoWatch_YieldsSnapshots(t *testing.T) { + withFastWatch(t, 10*time.Millisecond) + repo, cleanup := openTestDB(t) + defer cleanup() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + insertNode(t, repo, ctx, "node-1", "alpha") + + var snapshots [][]*model.Node + done := make(chan struct{}) + go func() { + defer close(done) + for snap := range repo.Watch(ctx) { + snapshots = append(snapshots, snap) + if len(snapshots) >= 15 { + cancel() + return + } + } + }() + + time.Sleep(100 * time.Millisecond) + insertNode(t, repo, context.Background(), "node-2", "beta") + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("watch did not complete within 2s") + } + + if len(snapshots) == 0 { + t.Fatal("expected at least one snapshot, got none") + } + if len(snapshots[0]) != 1 || snapshots[0][0].ID != "node-1" { + t.Errorf("first snapshot = %+v, want only node-1", snapshots[0]) + } + foundBoth := false + for _, snap := range snapshots[1:] { + ids := make(map[string]bool, len(snap)) + for _, n := range snap { + ids[n.ID] = true + } + if ids["node-1"] && ids["node-2"] { + foundBoth = true + break + } + } + if !foundBoth { + t.Errorf("no snapshot contained both nodes; snapshots=%v", snapshots) + } +} + +func TestNodeRepoWatch_ImmediateFirstYield(t *testing.T) { + withFastWatch(t, 200*time.Millisecond) + repo, cleanup := openTestDB(t) + defer cleanup() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + insertNode(t, repo, ctx, "node-immediate", "first") + + start := time.Now() + var firstSnap []*model.Node + got := make(chan struct{}) + go func() { + for snap := range repo.Watch(ctx) { + firstSnap = snap + close(got) + cancel() + return + } + }() + + select { + case <-got: + case <-time.After(100 * time.Millisecond): + t.Fatal("first yield took >100ms; expected immediate (G-002)") + } + + elapsed := time.Since(start) + if elapsed > 100*time.Millisecond { + t.Errorf("first yield took %v; expected immediate (G-002)", elapsed) + } + if len(firstSnap) != 1 || firstSnap[0].ID != "node-immediate" { + t.Errorf("first snapshot = %+v, want node-immediate", firstSnap) + } +} + +func TestNodeRepoWatch_StopsOnConsumerBreak(t *testing.T) { + withFastWatch(t, 10*time.Millisecond) + repo, cleanup := openTestDB(t) + defer cleanup() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + insertNode(t, repo, ctx, "node-break", "break") + + done := make(chan struct{}) + go func() { + defer close(done) + for range repo.Watch(ctx) { + break + } + }() + + select { + case <-done: + case <-time.After(500 * time.Millisecond): + t.Fatal("watch did not stop on consumer break within 500ms") + } +} + +func TestNodeRepoWatch_StopsOnCtxCancel(t *testing.T) { + withFastWatch(t, 10*time.Millisecond) + repo, cleanup := openTestDB(t) + defer cleanup() + + ctx, cancel := context.WithCancel(context.Background()) + + insertNode(t, repo, ctx, "node-cancel", "cancel") + + done := make(chan struct{}) + go func() { + defer close(done) + for range repo.Watch(ctx) { + } + }() + + time.Sleep(20 * time.Millisecond) + cancel() + + select { + case <-done: + case <-time.After(500 * time.Millisecond): + t.Fatal("watch did not stop on ctx cancel within 500ms") + } +}