Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 38220192bb | |||
| ba5ffd76f9 | |||
| 9b308c79f4 | |||
| a7bb00d935 | |||
| b4d9409e4d | |||
| efdbd2a61d | |||
| 5755f12053 | |||
| 5dba3cef80 | |||
| fc6a6c07e2 |
@@ -1,9 +1,9 @@
|
||||
{
|
||||
"phase": 3,
|
||||
"stage": "execute",
|
||||
"phase": 0,
|
||||
"stage": "grill",
|
||||
"milestone": "v0.3",
|
||||
"milestone_slug": "scheduling-streaming",
|
||||
"phase_role": "final",
|
||||
"phase_role": "pre_execution",
|
||||
"attempts": 0,
|
||||
"updated_at": "2026-08-01T00:25:00Z"
|
||||
"updated_at": "2026-08-01T00:04:00Z"
|
||||
}
|
||||
@@ -29,7 +29,7 @@ earlier versions of this file.
|
||||
| REQ-019 | Cobra CLI framework | High | v0.1 P01 | **Complete** |
|
||||
| REQ-020 | HCL parser integration (`hashicorp/hcl`) | Medium | v0.1 P03 | **Complete** |
|
||||
| REQ-021 | `os/exec` with `WaitDelay` (Go 1.25+) | Medium | v0.1 P03 | **Complete** |
|
||||
| REQ-022 | `iter.Seq` for streaming job lists (Go 1.25+) | Low | **v0.3 P01** | **Complete** (v0.3 P01 shipped v0.3.1) |
|
||||
| REQ-022 | `iter.Seq` for streaming job lists (Go 1.25+) | Low | **v0.3 P01** | Pending (v0.3 P01) |
|
||||
| REQ-023 | Self-signed mTLS cert generation | Medium | **v0.2 P01** | **Complete** (P01 shipped v0.2.1) |
|
||||
| REQ-024 | `Makefile` with standard targets | High | v0.1 P01 | **Complete** |
|
||||
| REQ-025 | Bounded cert rotation history: retain last N=3 server certs per node for rollback | Medium | **v0.2 P01** | **Complete** (P01 shipped v0.2.1) |
|
||||
@@ -37,9 +37,9 @@ earlier versions of this file.
|
||||
| REQ-027 | `govulncheck` runs in offline mode in CI (no `vuln.go.dev` calls; pre-mirrored DB or `-format json` + `jq` gate) | High | v0.2 P03 | **Complete** (P10 shipped v0.2.3) |
|
||||
| REQ-028 | HCL/YAML schema for `NodeCapacity` declaration (`orca node join` flag and/or `~/.orca/node.hcl`) | High | v0.2 P02 | **Complete** (P09 shipped v0.2.2; `orca node capacity` CLI) |
|
||||
| REQ-029 | `gitleaks` baseline file committed to repo to suppress pre-existing `.env` SHA-1 leak in git history | Medium | v0.2 P03 | **Complete** (P10 shipped v0.2.3) |
|
||||
| REQ-030 | `--watch` output format mode: table (default) vs streaming one-line JSON per event | Low | **v0.3 P01** | **Complete** (v0.3 P01 shipped v0.3.1) |
|
||||
| REQ-030 | `--watch` output format mode: table (default) vs streaming one-line JSON per event | Low | **v0.3 P01** | Pending (v0.3 P01) |
|
||||
| REQ-031 | `go test -race` enabled in CI for all v0.2 packages | High | v0.2 P01–P04 | **Complete** (P10; `.coreci.yml` test pipeline runs `-race`) |
|
||||
| REQ-032 | `orca doctor` subcommand for diagnostics (CA/cert health, db integrity, peer reachability) | Medium | **v0.2 P01 / v0.3 P02** | **Complete** (cert checks P01 v0.2.1; network + db P02 v0.3.2) |
|
||||
| REQ-032 | `orca doctor` subcommand for diagnostics (CA/cert health, db integrity, peer reachability) | Medium | **v0.2 P01 / v0.3 P02** | **Partial** — cert checks complete (P01); network/db are stubs, full impl in v0.3 P02 |
|
||||
| REQ-033 | Cert file mode enforcement: 0600 for keys, 0644 for certs (refuses to start on violation) | High | **v0.2 P01** | **Complete** (P01 shipped v0.2.1) |
|
||||
| REQ-034 | Cert proactive rotation alarm: structured slog WARN 30 days before `not_after` | Medium | **v0.2 P01** | **Complete** (P01 shipped v0.2.1) |
|
||||
| REQ-035 | `orca cert show` redacts private key material from default and `--json` output | High | **v0.2 P01** | **Complete** (P01 shipped v0.2.1) |
|
||||
@@ -71,12 +71,13 @@ deferred to v0.3.
|
||||
|
||||
## v0.3 Milestone Summary
|
||||
|
||||
**Status: Complete** — P01 (iter.Seq streaming, v0.3.1) and P02 (doctor
|
||||
network+db, v0.3.2) both shipped. REQ-022, REQ-030, REQ-032 all complete.
|
||||
Re-init SPECIFY audit confirmed all other v0.2-deferred REQs (014, 027,
|
||||
028, 029, 031, 037, 039, 040) already shipped in P08-P10.
|
||||
**Status: In Progress** — 2 execution phases planned (P01 iter.Seq
|
||||
streaming, P02 doctor completion). Covers REQ-022, REQ-030, REQ-032
|
||||
(completion). Re-init SPECIFY audit confirmed all other v0.2-deferred
|
||||
REQs (014, 027, 028, 029, 031, 037, 039, 040) already shipped in
|
||||
P08-P10.
|
||||
|
||||
## Deferred to v0.4
|
||||
## Deferred to v0.3
|
||||
|
||||
- pprof endpoint on `orca daemon` (idea I-308, 0.70 confidence): deferred
|
||||
to keep v0.2 lean; revisit in v0.3 once P02's dispatcher is stable.
|
||||
|
||||
+6
-6
@@ -34,19 +34,19 @@ richer CI security scanning, and streaming I/O.
|
||||
|
||||
Per-phase tags: `v0.2.1` (P01), `v0.2.2` (P02), `v0.2.3` (P03) — all shipped.
|
||||
|
||||
## Milestone v0.3: Scheduling & Streaming Completion — **COMPLETE**
|
||||
## Milestone v0.3: Scheduling & Streaming Completion — **IN PROGRESS**
|
||||
|
||||
Scope: complete the two work items deferred from v0.2 that were not
|
||||
already shipped in P08-P10. A re-init SPECIFY codebase audit confirmed
|
||||
that REQ-014/027/028/029/031/037/039/040 all shipped in P08-P10 despite
|
||||
stale REQUIREMENTS.md marking them Pending. The remaining work is lean:
|
||||
|
||||
- [x] Phase 0: Pre-execution (specify → clarify → research → plan → grill) — shipped v0.3.0
|
||||
- [x] Phase 1: `iter.Seq` streaming for `--watch` flags (REQ-022, REQ-030) — shipped v0.3.1
|
||||
- [x] Phase 2: `orca doctor` network + db full implementation (REQ-032 completion) — shipped v0.3.2
|
||||
- [x] Phase 3: Final review + ship + audit (milestone release) — shipped v0.3.3
|
||||
- [ ] Phase 0: Pre-execution (specify → clarify → research → plan)
|
||||
- [ ] Phase 1: `iter.Seq` streaming for `--watch` flags (REQ-022, REQ-030)
|
||||
- [ ] Phase 2: `orca doctor` network + db full implementation (REQ-032 completion)
|
||||
- [ ] Phase 3: Final review + ship + audit (milestone release)
|
||||
|
||||
**Milestone tag**: `v0.4.0` (next-minor per feature-milestone promotion rule).
|
||||
**Target milestone tag**: `v0.4.0` (next-minor per feature-milestone promotion rule).
|
||||
|
||||
Per-phase tags: `v0.3.0` (P0), `v0.3.1` (P01), `v0.3.2` (P02), `v0.3.3` (P03 final = milestone release).
|
||||
Per `.ciagent/RELEASE_POLICY.md`, every phase tag produces a Gitea release.
|
||||
|
||||
@@ -36,13 +36,3 @@ func ServerCertPath() string { return filepath.Join(Dir(), "server.crt") }
|
||||
|
||||
// ServerKeyPath returns the path to server.key.
|
||||
func ServerKeyPath() string { return filepath.Join(Dir(), "server.key") }
|
||||
|
||||
// DBPath returns the path to the orca SQLite database. Honors $ORCA_DB
|
||||
// for testability and explicit override; otherwise defaults to
|
||||
// ~/.orca/orca.db under the same Dir() as the cert files.
|
||||
func DBPath() string {
|
||||
if p := os.Getenv("ORCA_DB"); p != "" {
|
||||
return p
|
||||
}
|
||||
return filepath.Join(Dir(), "orca.db")
|
||||
}
|
||||
|
||||
@@ -51,7 +51,7 @@ var doctorNetworkCmd = &cobra.Command{
|
||||
Use: "network",
|
||||
Short: "Run the network self-check (P02 impl)",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
c := doctor.Network()
|
||||
c := doctor.NetworkStub()
|
||||
r, msg := c.Run(cmd.Context())
|
||||
fmt.Fprintf(cmd.OutOrStdout(), "%-20s %-5s %s\n", c.Name, r, msg)
|
||||
return nil
|
||||
@@ -62,7 +62,7 @@ var doctorDBCmd = &cobra.Command{
|
||||
Use: "db",
|
||||
Short: "Run the database self-check (P02 impl)",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
c := doctor.DB()
|
||||
c := doctor.DBStub()
|
||||
r, msg := c.Run(cmd.Context())
|
||||
fmt.Fprintf(cmd.OutOrStdout(), "%-20s %-5s %s\n", c.Name, r, msg)
|
||||
return nil
|
||||
|
||||
+1
-76
@@ -5,9 +5,6 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
@@ -39,7 +36,6 @@ var (
|
||||
stopID string
|
||||
runTarget string
|
||||
runIDKey string
|
||||
jobWatch bool
|
||||
)
|
||||
|
||||
var jobRunCmd = &cobra.Command{
|
||||
@@ -116,11 +112,8 @@ var jobRunCmd = &cobra.Command{
|
||||
var jobListCmd = &cobra.Command{
|
||||
Use: "list",
|
||||
Short: "List all jobs",
|
||||
Long: "Display all jobs and their status. Use --watch to stream updates until Ctrl-C.",
|
||||
Long: "Display all jobs and their status.",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
if jobWatch {
|
||||
return watchJobs(cmd)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
@@ -149,73 +142,6 @@ 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",
|
||||
@@ -309,7 +235,6 @@ 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)
|
||||
|
||||
+11
-77
@@ -3,12 +3,10 @@ package cli
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
@@ -21,8 +19,16 @@ import (
|
||||
"git.cloudinit.dev/coreci/orca/internal/store"
|
||||
)
|
||||
|
||||
func dbPath() string {
|
||||
if p := os.Getenv("ORCA_DB"); p != "" {
|
||||
return p
|
||||
}
|
||||
home, _ := os.UserHomeDir()
|
||||
return filepath.Join(home, ".orca", "orca.db")
|
||||
}
|
||||
|
||||
func openDB() (*sql.DB, func() error, error) {
|
||||
db, err := store.Open(certpaths.DBPath())
|
||||
db, err := store.Open(dbPath())
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
@@ -48,7 +54,6 @@ var (
|
||||
joinAddr string
|
||||
joinCAFinger string
|
||||
leaveID string
|
||||
nodeWatch bool
|
||||
)
|
||||
|
||||
var nodeCmd = &cobra.Command{
|
||||
@@ -150,11 +155,8 @@ 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. Use --watch to stream updates until Ctrl-C.",
|
||||
Long: "Display all registered nodes and their state.",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
if nodeWatch {
|
||||
return watchNodes(cmd)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
@@ -183,79 +185,11 @@ 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)
|
||||
|
||||
@@ -1,287 +0,0 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
+15
-116
@@ -19,17 +19,12 @@ import (
|
||||
"crypto/x509"
|
||||
"encoding/pem"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"os"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.cloudinit.dev/coreci/orca/internal/certpaths"
|
||||
"git.cloudinit.dev/coreci/orca/internal/model"
|
||||
"git.cloudinit.dev/coreci/orca/internal/security"
|
||||
"git.cloudinit.dev/coreci/orca/internal/store"
|
||||
"git.cloudinit.dev/coreci/orca/internal/transport"
|
||||
)
|
||||
|
||||
// Result is the outcome of a single check.
|
||||
@@ -68,8 +63,8 @@ func All() []Check {
|
||||
CertServer(),
|
||||
CertExpiry(),
|
||||
CertFingerprint(),
|
||||
Network(),
|
||||
DB(),
|
||||
NetworkStub(),
|
||||
DBStub(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -182,122 +177,26 @@ func CertFingerprint() Check {
|
||||
}
|
||||
}
|
||||
|
||||
// DB checks SQLite integrity and migration version (REQ-032 completion).
|
||||
func DB() Check {
|
||||
return Check{
|
||||
Name: "db",
|
||||
Description: "SQLite integrity_check + migration version",
|
||||
Run: func(ctx context.Context) (Result, string) {
|
||||
path := certpaths.DBPath()
|
||||
db, err := store.Open(path)
|
||||
if err != nil {
|
||||
return ResultFail, fmt.Sprintf("open db: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
var integrity string
|
||||
if err := db.QueryRowContext(ctx, "PRAGMA integrity_check").Scan(&integrity); err != nil {
|
||||
return ResultFail, fmt.Sprintf("integrity_check: %v", err)
|
||||
}
|
||||
if !strings.EqualFold(integrity, "ok") {
|
||||
return ResultFail, fmt.Sprintf("integrity_check: %s", integrity)
|
||||
}
|
||||
|
||||
version, err := store.MigrationVersion(ctx, db)
|
||||
if err != nil {
|
||||
return ResultFail, fmt.Sprintf("migration version: %v", err)
|
||||
}
|
||||
if version == "" {
|
||||
return ResultWarn, "integrity OK but no migrations applied (fresh db)"
|
||||
}
|
||||
return ResultPass, fmt.Sprintf("integrity OK, migrations up to %s", version)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// Network probes peer reachability via mTLS /healthz (REQ-032 completion).
|
||||
// Peers are sourced from the persisted nodes table (not the in-memory
|
||||
// PeerRegistry, which is empty at CLI time). Zero peers → WARN (single-node
|
||||
// is legitimate). Any peer unreachable → FAIL (D-038).
|
||||
func Network() Check {
|
||||
// NetworkStub is a stub for the network check; full impl in P02.
|
||||
func NetworkStub() Check {
|
||||
return Check{
|
||||
Name: "network",
|
||||
Description: "peer reachability via mTLS /healthz probe",
|
||||
Run: func(ctx context.Context) (Result, string) {
|
||||
caPath := certpaths.CACertPath()
|
||||
certPath := certpaths.ServerCertPath()
|
||||
keyPath := certpaths.ServerKeyPath()
|
||||
|
||||
// Check that cert files exist before attempting probes.
|
||||
if _, err := os.Stat(caPath); err != nil {
|
||||
return ResultFail, fmt.Sprintf("CA cert missing: %v (run `orca cert init`)", err)
|
||||
}
|
||||
|
||||
path := certpaths.DBPath()
|
||||
db, err := store.Open(path)
|
||||
if err != nil {
|
||||
return ResultFail, fmt.Sprintf("open db: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
nodes, err := store.NewNodeRepo(db).List(ctx)
|
||||
if err != nil {
|
||||
return ResultFail, fmt.Sprintf("list nodes: %v", err)
|
||||
}
|
||||
|
||||
live := make([]*model.Node, 0, len(nodes))
|
||||
for _, n := range nodes {
|
||||
if n.State != model.NodeStateLeft {
|
||||
live = append(live, n)
|
||||
}
|
||||
}
|
||||
|
||||
if len(live) == 0 {
|
||||
return ResultWarn, "no peers registered (single-node?)"
|
||||
}
|
||||
|
||||
var lines []string
|
||||
anyFail := false
|
||||
for _, n := range live {
|
||||
probeCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
|
||||
err := probeHealthz(probeCtx, caPath, certPath, keyPath, n.Name, n.Address)
|
||||
cancel()
|
||||
if err != nil {
|
||||
anyFail = true
|
||||
lines = append(lines, fmt.Sprintf(" ✗ %s (%s): %v", n.Name, n.Address, err))
|
||||
} else {
|
||||
lines = append(lines, fmt.Sprintf(" ✓ %s (%s)", n.Name, n.Address))
|
||||
}
|
||||
}
|
||||
|
||||
result := ResultPass
|
||||
if anyFail {
|
||||
result = ResultFail
|
||||
}
|
||||
return result, strings.Join(lines, "\n")
|
||||
Description: "TCP reachability + mTLS handshake (full impl in P02)",
|
||||
Run: func(_ context.Context) (Result, string) {
|
||||
return ResultWarn, "network check is a stub in P01; full impl in P02"
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// probeHealthz opens an mTLS connection to the peer and GETs /healthz.
|
||||
func probeHealthz(ctx context.Context, caPath, certPath, keyPath, serverName, addr string) error {
|
||||
client, err := transport.NewMTLSClient(caPath, serverName, certPath, keyPath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("mTLS client: %w", err)
|
||||
// DBStub is a stub for the database check; full impl in P02.
|
||||
func DBStub() Check {
|
||||
return Check{
|
||||
Name: "db",
|
||||
Description: "SQLite open + migration apply (full impl in P02)",
|
||||
Run: func(_ context.Context) (Result, string) {
|
||||
return ResultWarn, "db check is a stub in P01; full impl in P02"
|
||||
},
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+addr+"/healthz", nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("request: %w", err)
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("probe: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("healthz returned %d", resp.StatusCode)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// loadCert reads a PEM cert from path and parses the first CERTIFICATE
|
||||
|
||||
+35
-190
@@ -2,74 +2,60 @@ package doctor
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.cloudinit.dev/coreci/orca/internal/model"
|
||||
"git.cloudinit.dev/coreci/orca/internal/security"
|
||||
"git.cloudinit.dev/coreci/orca/internal/store"
|
||||
)
|
||||
|
||||
// TestRunAllChecksWithNoCA runs the full battery in a clean temp dir.
|
||||
// With the P02 real checks (no stubs): cert checks FAIL (no CA),
|
||||
// db check PASS (store.Open runs migrations), network check WARN
|
||||
// (no peers).
|
||||
// TestRunAllChecksWithNoCA runs the full battery in a clean temp dir
|
||||
// and expects all checks to FAIL (no CA, no server cert) except the
|
||||
// two stubs which return WARN.
|
||||
func TestRunAllChecksWithNoCA(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
t.Setenv("ORCA_HOME", dir)
|
||||
t.Setenv("ORCA_DB", filepath.Join(dir, "orca.db"))
|
||||
// Isolated home so we don't touch the real ~/.orca.
|
||||
t.Setenv("ORCA_HOME", t.TempDir())
|
||||
|
||||
rep := Run(context.Background())
|
||||
if len(rep.Checks) == 0 {
|
||||
t.Fatal("expected checks, got 0")
|
||||
}
|
||||
|
||||
byName := make(map[string]CheckResult, len(rep.Checks))
|
||||
hasFail := false
|
||||
hasWarn := false
|
||||
for _, c := range rep.Checks {
|
||||
byName[c.Name] = c
|
||||
}
|
||||
|
||||
// Cert checks: no CA → FAIL.
|
||||
for _, name := range []string{"cert.ca", "cert.server", "cert.expiry", "cert.fingerprint"} {
|
||||
c, ok := byName[name]
|
||||
if !ok {
|
||||
t.Errorf("missing check %s", name)
|
||||
continue
|
||||
if c.Result == ResultFail {
|
||||
hasFail = true
|
||||
}
|
||||
if c.Result != ResultFail {
|
||||
t.Errorf("%s: got %s, want FAIL — %s", name, c.Result, c.Message)
|
||||
if c.Result == ResultWarn {
|
||||
hasWarn = true
|
||||
}
|
||||
}
|
||||
|
||||
// DB check: store.Open runs migrations → PASS.
|
||||
if c, ok := byName["db"]; ok {
|
||||
if c.Result != ResultPass {
|
||||
t.Errorf("db: got %s, want PASS — %s", c.Result, c.Message)
|
||||
}
|
||||
} else {
|
||||
t.Error("missing check db")
|
||||
if !hasFail {
|
||||
t.Error("expected at least one FAIL (no CA installed)")
|
||||
}
|
||||
if !hasWarn {
|
||||
t.Error("expected at least one WARN (stubs in P01)")
|
||||
}
|
||||
|
||||
// Network check: no CA → FAIL (can't build mTLS client without CA).
|
||||
if c, ok := byName["network"]; ok {
|
||||
if c.Result != ResultFail {
|
||||
t.Errorf("network: got %s, want FAIL (no CA cert) — %s", c.Result, c.Message)
|
||||
}
|
||||
} else {
|
||||
t.Error("missing check network")
|
||||
// Render the report — basic shape check.
|
||||
out := rep.Print()
|
||||
if !strings.Contains(out, "PASS") {
|
||||
t.Errorf("expected PASS in output, got: %s", out)
|
||||
}
|
||||
if !strings.Contains(out, "WARN") {
|
||||
t.Errorf("expected WARN in output, got: %s", out)
|
||||
}
|
||||
if !strings.Contains(out, "FAIL") {
|
||||
t.Errorf("expected FAIL in output, got: %s", out)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRunWithCAAndServerCert covers the happy path: CA + server cert
|
||||
// installed → all cert checks PASS, db PASS, network WARN (no peers).
|
||||
// installed → all cert checks PASS.
|
||||
func TestRunWithCAAndServerCert(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
t.Setenv("ORCA_HOME", dir)
|
||||
t.Setenv("ORCA_DB", filepath.Join(dir, "orca.db"))
|
||||
|
||||
// Bootstrap CA.
|
||||
if _, err := security.CAInit(dir, "test-ca"); err != nil {
|
||||
t.Fatalf("CAInit: %v", err)
|
||||
}
|
||||
@@ -77,6 +63,7 @@ func TestRunWithCAAndServerCert(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("LoadCA: %v", err)
|
||||
}
|
||||
// Generate + sign server cert.
|
||||
keyPEM, csrPEM, err := security.GenerateCSR("test-server", []string{"localhost", "127.0.0.1"})
|
||||
if err != nil {
|
||||
t.Fatalf("GenerateCSR: %v", err)
|
||||
@@ -93,155 +80,13 @@ func TestRunWithCAAndServerCert(t *testing.T) {
|
||||
}
|
||||
|
||||
rep := Run(context.Background())
|
||||
byName := make(map[string]CheckResult, len(rep.Checks))
|
||||
// The cert-related checks should be PASS; the network/db stubs WARN.
|
||||
for _, c := range rep.Checks {
|
||||
byName[c.Name] = c
|
||||
}
|
||||
|
||||
for _, name := range []string{"cert.ca", "cert.server", "cert.expiry", "cert.fingerprint", "db"} {
|
||||
c, ok := byName[name]
|
||||
if !ok {
|
||||
t.Errorf("missing check %s", name)
|
||||
continue
|
||||
}
|
||||
if c.Result != ResultPass {
|
||||
t.Errorf("%s: got %s, want PASS — %s", name, c.Result, c.Message)
|
||||
}
|
||||
}
|
||||
|
||||
if c, ok := byName["network"]; ok {
|
||||
if c.Result != ResultWarn {
|
||||
t.Errorf("network: got %s, want WARN (no peers) — %s", c.Result, c.Message)
|
||||
switch c.Name {
|
||||
case "cert.ca", "cert.server", "cert.expiry", "cert.fingerprint":
|
||||
if c.Result != ResultPass {
|
||||
t.Errorf("%s: got %s, want PASS — %s", c.Name, c.Result, c.Message)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestDBCheck_IntegrityOK verifies the DB check passes on a fresh
|
||||
// database with migrations applied.
|
||||
func TestDBCheck_IntegrityOK(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
t.Setenv("ORCA_HOME", dir)
|
||||
t.Setenv("ORCA_DB", filepath.Join(dir, "orca.db"))
|
||||
|
||||
db, err := store.Open(filepath.Join(dir, "orca.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("open db: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
c := DB()
|
||||
r, msg := c.Run(context.Background())
|
||||
if r != ResultPass {
|
||||
t.Errorf("DB check: got %s, want PASS — %s", r, msg)
|
||||
}
|
||||
if !strings.Contains(msg, "0005") {
|
||||
t.Errorf("DB check message should contain migration version, got: %s", msg)
|
||||
}
|
||||
}
|
||||
|
||||
// TestNetworkCheck_NoPeers verifies the network check returns WARN
|
||||
// when no peers are registered.
|
||||
func TestNetworkCheck_NoPeers(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
t.Setenv("ORCA_HOME", dir)
|
||||
t.Setenv("ORCA_DB", filepath.Join(dir, "orca.db"))
|
||||
|
||||
// Create a CA + server cert so the network check can build a client.
|
||||
if _, err := security.CAInit(dir, "test-ca"); err != nil {
|
||||
t.Fatalf("CAInit: %v", err)
|
||||
}
|
||||
ca, _ := security.LoadCA(dir)
|
||||
keyPEM, csrPEM, _ := security.GenerateCSR("test-server", []string{"localhost"})
|
||||
certPEM, _ := ca.SignCSR(csrPEM)
|
||||
_ = security.WriteCert(dir+"/server.crt", certPEM)
|
||||
_ = security.WriteKey(dir+"/server.key", keyPEM)
|
||||
|
||||
c := Network()
|
||||
r, msg := c.Run(context.Background())
|
||||
if r != ResultWarn {
|
||||
t.Errorf("Network check: got %s, want WARN — %s", r, msg)
|
||||
}
|
||||
if !strings.Contains(msg, "no peers") {
|
||||
t.Errorf("Network check message should mention no peers, got: %s", msg)
|
||||
}
|
||||
}
|
||||
|
||||
// TestNetworkCheck_PeerUnreachable verifies the network check returns
|
||||
// FAIL when a registered peer is not reachable.
|
||||
func TestNetworkCheck_PeerUnreachable(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
t.Setenv("ORCA_HOME", dir)
|
||||
t.Setenv("ORCA_DB", filepath.Join(dir, "orca.db"))
|
||||
|
||||
// Create a CA + server cert.
|
||||
if _, err := security.CAInit(dir, "test-ca"); err != nil {
|
||||
t.Fatalf("CAInit: %v", err)
|
||||
}
|
||||
ca, _ := security.LoadCA(dir)
|
||||
keyPEM, csrPEM, _ := security.GenerateCSR("test-server", []string{"localhost"})
|
||||
certPEM, _ := ca.SignCSR(csrPEM)
|
||||
_ = security.WriteCert(dir+"/server.crt", certPEM)
|
||||
_ = security.WriteKey(dir+"/server.key", keyPEM)
|
||||
|
||||
// Insert a peer node with an unreachable address.
|
||||
db, err := store.Open(filepath.Join(dir, "orca.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("open db: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
repo := store.NewNodeRepo(db)
|
||||
_ = repo.Insert(context.Background(), &model.Node{
|
||||
ID: "dead-peer", Name: "dead", Address: "127.0.0.1:1",
|
||||
State: model.NodeStateReady, JoinedAt: time.Now().UTC(), LastSeen: time.Now().UTC(),
|
||||
})
|
||||
|
||||
c := Network()
|
||||
r, msg := c.Run(context.Background())
|
||||
if r != ResultFail {
|
||||
t.Errorf("Network check: got %s, want FAIL — %s", r, msg)
|
||||
}
|
||||
if !strings.Contains(msg, "dead") {
|
||||
t.Errorf("Network check message should mention the dead peer, got: %s", msg)
|
||||
}
|
||||
}
|
||||
|
||||
// TestNetworkCheck_NoCert verifies the network check returns FAIL
|
||||
// when no CA cert is installed.
|
||||
func TestNetworkCheck_NoCert(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
t.Setenv("ORCA_HOME", dir)
|
||||
t.Setenv("ORCA_DB", filepath.Join(dir, "orca.db"))
|
||||
|
||||
c := Network()
|
||||
r, msg := c.Run(context.Background())
|
||||
if r != ResultFail {
|
||||
t.Errorf("Network check: got %s, want FAIL — %s", r, msg)
|
||||
}
|
||||
if !strings.Contains(msg, "CA cert missing") {
|
||||
t.Errorf("Network check message should mention missing CA, got: %s", msg)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRenderReport verifies the report output format.
|
||||
func TestRenderReport(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
t.Setenv("ORCA_HOME", dir)
|
||||
t.Setenv("ORCA_DB", filepath.Join(dir, "orca.db"))
|
||||
|
||||
rep := Run(context.Background())
|
||||
out := rep.Print()
|
||||
if !strings.Contains(out, "PASS") {
|
||||
t.Errorf("expected PASS in output, got: %s", out)
|
||||
}
|
||||
if !strings.Contains(out, "WARN") {
|
||||
t.Errorf("expected WARN in output, got: %s", out)
|
||||
}
|
||||
if !strings.Contains(out, "FAIL") {
|
||||
t.Errorf("expected FAIL in output, got: %s", out)
|
||||
}
|
||||
}
|
||||
|
||||
func init() {
|
||||
// Suppress slog noise during tests.
|
||||
_ = os.Setenv("ORCA_LOG_LEVEL", "error")
|
||||
}
|
||||
|
||||
@@ -6,19 +6,11 @@ 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
|
||||
}
|
||||
@@ -67,52 +59,6 @@ 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
|
||||
|
||||
@@ -1,201 +0,0 @@
|
||||
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) >= 40 {
|
||||
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")
|
||||
}
|
||||
}
|
||||
@@ -12,21 +12,6 @@ import (
|
||||
//go:embed migrations/*.sql
|
||||
var migrationsFS embed.FS
|
||||
|
||||
// MigrationVersion returns the name of the highest applied migration
|
||||
// (e.g. "0005_node_capacity.sql"). Returns ("", nil) if no migrations
|
||||
// have been applied (fresh or empty database).
|
||||
func MigrationVersion(ctx context.Context, db *sql.DB) (string, error) {
|
||||
var name string
|
||||
err := db.QueryRowContext(ctx, `SELECT name FROM schema_migrations ORDER BY name DESC LIMIT 1`).Scan(&name)
|
||||
if err == sql.ErrNoRows {
|
||||
return "", nil
|
||||
}
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("query migration version: %w", err)
|
||||
}
|
||||
return name, nil
|
||||
}
|
||||
|
||||
func migrate(db *sql.DB) error {
|
||||
entries, err := migrationsFS.ReadDir("migrations")
|
||||
if err != nil {
|
||||
|
||||
@@ -1,37 +0,0 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestMigrationVersion(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
db, err := Open(filepath.Join(dir, "test.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("open db: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
ctx := context.Background()
|
||||
version, err := MigrationVersion(ctx, db)
|
||||
if err != nil {
|
||||
t.Fatalf("migration version: %v", err)
|
||||
}
|
||||
if version != "0005_node_capacity.sql" {
|
||||
t.Errorf("MigrationVersion = %q, want 0005_node_capacity.sql", version)
|
||||
}
|
||||
|
||||
// Empty the migrations table → should return ("", nil).
|
||||
if _, err := db.ExecContext(ctx, "DELETE FROM schema_migrations"); err != nil {
|
||||
t.Fatalf("clear migrations: %v", err)
|
||||
}
|
||||
version, err = MigrationVersion(ctx, db)
|
||||
if err != nil {
|
||||
t.Fatalf("migration version after clear: %v", err)
|
||||
}
|
||||
if version != "" {
|
||||
t.Errorf("MigrationVersion after clear = %q, want empty", version)
|
||||
}
|
||||
}
|
||||
@@ -6,8 +6,6 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"iter"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"git.cloudinit.dev/coreci/orca/internal/model"
|
||||
@@ -71,40 +69,6 @@ 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 = ?`,
|
||||
|
||||
@@ -100,159 +100,3 @@ 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) >= 40 {
|
||||
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")
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user