diff --git a/docs/cli.md b/docs/cli.md index 1f715405..6640e86d 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -480,7 +480,13 @@ this time that time out enter stop cleanup; existing runs are not stopped or res wait failed. The cleanup names each run it started by `run_id`: a run another controller started for the same service while the wait was still running is a different run and is left alone. A daemon that does not announce `groups.kill.expect_runs` cannot scope a stop to named runs, so it is not sent a -name-only cleanup: the command fails and says the timed-out runs are still running. The `started` / +name-only cleanup: the command fails and says the timed-out runs are still running. Readiness is +attributed by the same identity: when the service row's live run is not the run this wait is about, the +wait ends immediately with state `failed` and reason `run_replaced`, because the replacement's listener +and health are its own evidence, not the original run's; nothing is stopped then, since the replacement +is not this invocation's run. A row that names no run cannot confirm a named one: a listener not +attributed to the run keeps the wait going instead of counting as its readiness. A chunk with no +`run_id` at all carries no identity to compare, so the service row is read as before. The `started` / `skipped` summary describes the start action, and readiness failures appear separately in `errors` and in each service's `state`. For runs newly started this time that were cleaned up, the engine writes the same `ready_timeout` into the service's last exit facts, so diff --git a/docs/cli.zh.md b/docs/cli.zh.md index f18b5e32..45e187da 100644 --- a/docs/cli.zh.md +++ b/docs/cli.zh.md @@ -400,7 +400,11 @@ cwd 中执行，按 `cmd:` 的规则拆成 argv（不经过 shell），支持 `$ 已经存在的运行不会因这次等待失败而被停止或重启。清理按 `run_id` 只停止本次启动的那个运行： 等待期间另一个控制者为同一服务启动的新运行是另一个运行，不会被停掉。daemon 若未声明 `groups.kill.expect_runs`，就无法把停止限定到指定运行，因此不会收到只带服务名的清理： -命令失败，并说明超时的运行仍在运行。`started` / `skipped` 汇总描述启动动作， +命令失败，并说明超时的运行仍在运行。就绪判定同样按运行身份归属：状态行里的运行不是这次等待所指的 +运行时就立即以状态 `failed`、原因 `run_replaced` 结束，因为替换运行的监听与健康是它自己的证据， +不能算作原运行的；此时不停止任何运行，替换运行不是本次调用启动的。状态行没有指明运行时也无法确认具名运行： +无法归属到该运行的监听只会让等待继续，不会被算作它的就绪。完全没有 `run_id` 的 chunk +没有可比对的身份，仍按服务行判断。`started` / `skipped` 汇总描述启动动作， 就绪判定失败另外出现在 `errors` 与逐服务 `state` 中。对本次新启动且被清理的运行， 引擎把同一个 `ready_timeout` 写入服务的最后退出事实，因此 `status`、桌面服务行、attention 和事件流 不会把它显示成普通的手动停止。健康事实带有 `status`、`code`、`reason` 与 `observed_at`， diff --git a/engine/internal/cmd/up_ready.go b/engine/internal/cmd/up_ready.go index 70177569..bd3793ad 100644 --- a/engine/internal/cmd/up_ready.go +++ b/engine/internal/cmd/up_ready.go @@ -14,6 +14,11 @@ import ( "github.com/sheathedsharp/option-berth/engine/internal/state" ) +// startRows reads the state a readiness wait decides on. Production reads the +// same snapshot status exposes; a test replaces it so a readiness rule can be +// checked without a daemon or the host's listener table. +var startRows = groupRowsObserved + // startTimeout is one service this wait gave up on: the name it was started // under and the run this invocation started. The run id is what authorizes the // cleanup — a service name alone would also match a run another controller @@ -62,7 +67,7 @@ func waitForStartReady(ctx context.Context, params rpc.GroupsStartParams, cfg *g fallbackSince := waitStart.UTC().Truncate(time.Second) var timedOut []startTimeout for { - pp, gg, obs, err := groupRowsObserved(ctx) + pp, gg, obs, err := startRows(ctx) if obs != nil && obs.State != string(ports.ObservationObservable) { // Waiting the full timeout would end by stopping the run it just // started. "Cannot see" is a different outcome from "not ready". @@ -96,7 +101,17 @@ func waitForStartReady(ctx context.Context, params rpc.GroupsStartParams, cfg *g if group == nil || svc == nil { continue } - if ready, _ := serviceReady(*group, portsNow, name); ready { + // Readiness is the selected run's, not the service name's. A + // row that names another run is a replacement, and a row that + // names none cannot confirm a named one: its listener and its + // health answer for whatever it does name. + if replacement, ok := replacementRun(selection, svc); ok { + markStartChunkFailed(chunks, summary, name, "run_replaced", + startReplacedMessage(selection, replacement), nil) + delete(selected, name) + continue + } + if ready, _ := serviceReady(*group, portsNow, name); ready && startRowConfirms(selection, svc) { markStartChunkReady(chunks, name) delete(selected, name) continue @@ -198,6 +213,41 @@ func startTimeoutMessage(selection startSelection, name string, timeout time.Dur selection.budget, name) } +// replacementRun names the live run that took this selection's place. A row +// for the same service that names another run answers for that run: its +// listener and its health are evidence about the replacement, never about the +// run this wait is about. Only two known run ids can disagree — a missing id +// on either side is no identity evidence, and the wait keeps reading the row. +func replacementRun(selection startSelection, svc *state.Service) (string, bool) { + if selection.runID == "" || svc == nil || !svc.Running || svc.RunID == nil { + return "", false + } + if *svc.RunID == "" || *svc.RunID == selection.runID { + return "", false + } + return *svc.RunID, true +} + +// startRowConfirms reports whether the service row is evidence about the +// selected run. A row that names no run cannot confirm a named one: whatever +// is listening there is some process on the service's address, not this run's +// readiness. A selection with no run id has nothing to compare and keeps +// reading the row. +func startRowConfirms(selection startSelection, svc *state.Service) bool { + if selection.runID == "" { + return true + } + return svc != nil && svc.RunID != nil && *svc.RunID == selection.runID +} + +// startReplacedMessage states what ended the wait. It does not call the +// replacement unhealthy: the point is that it is not the run this wait was +// asked to confirm, so it cannot answer for it. +func startReplacedMessage(selection startSelection, replacement string) string { + return fmt.Sprintf("the run this wait is about (%s) was replaced by %s while waiting", + selection.runID, replacement) +} + // startChunkReadyTimeout is the readiness budget the daemon published for one // service. A daemon predating ready_timeout omits it; the client then falls // back to the same default the manifest uses. diff --git a/engine/internal/cmd/up_ready_test.go b/engine/internal/cmd/up_ready_test.go new file mode 100644 index 00000000..0793ac20 --- /dev/null +++ b/engine/internal/cmd/up_ready_test.go @@ -0,0 +1,181 @@ +package cmd + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/sheathedsharp/option-berth/engine/internal/daemon/rpc" + "github.com/sheathedsharp/option-berth/engine/internal/groups" + "github.com/sheathedsharp/option-berth/engine/internal/ports" + "github.com/sheathedsharp/option-berth/engine/internal/state" +) + +// startRowsFrom replaces the wait's observation with one group row set, so a +// readiness rule is decided on rows a test controls instead of a daemon or the +// host's listener table. +func startRowsFrom(t *testing.T, groups ...state.Group) { + t.Helper() + prev := startRows + startRows = func(context.Context) ([]ports.ListeningPort, []state.Group, *rpc.ObservationStatus, error) { + return nil, groups, &rpc.ObservationStatus{State: string(ports.ObservationObservable)}, nil + } + t.Cleanup(func() { startRows = prev }) +} + +// startManifest writes one project manifest and returns its path. +func startManifest(t *testing.T, services string) string { + t.Helper() + dir := t.TempDir() + path := filepath.Join(dir, groups.ConfigName) + if err := os.WriteFile(path, []byte("name: demo\nservices:\n"+services), 0o600); err != nil { + t.Fatal(err) + } + return path +} + +// TestWaitForStartReadyCreditsOnlyTheRunItIsAbout: the wait is about a run, not +// about whatever answers to the service's name. R1 was started here and is +// still waiting; another controller restarts the same service as R2, and R2 +// becomes healthy. R2's health is evidence about R2 — reading it as R1's +// readiness reports a service that was never confirmed, and the run that is +// actually healthy is not this invocation's to stop. +func TestWaitForStartReadyCreditsOnlyTheRunItIsAbout(t *testing.T) { + port, healthz := 19001, "/healthz" + for _, tc := range []struct { + name string + chunk rpc.GroupsStartChunk + service state.Service + }{ + { + name: "a healthy replacement listener", + chunk: rpc.GroupsStartChunk{Service: "api", State: "started", RunID: "run-r1"}, + service: state.Service{PortActual: &port, Health: &healthz, + HealthStatus: &state.Health{Status: state.HealthOK}}, + }, + { + name: "a healthy replacement worker", + chunk: rpc.GroupsStartChunk{Service: "api", State: "started", RunID: "run-r1"}, + service: state.Service{}, + }, + { + name: "a skipped run the chunk named", + chunk: rpc.GroupsStartChunk{Service: "api", State: "skipped", Skipped: true, RunID: "run-r1"}, + service: state.Service{PortActual: &port, Health: &healthz, + HealthStatus: &state.Health{Status: state.HealthOK}}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + path := startManifest(t, " - name: api\n cmd: echo api\n") + replacement := "run-r2" + svc := tc.service + svc.Name = "api" + svc.Running = true + svc.RunID = &replacement + startRowsFrom(t, state.Group{Name: "demo", ConfigPath: &path, Services: []state.Service{svc}}) + + chunk := tc.chunk + chunk.ReadyTimeout = "250ms" + chunks := []rpc.GroupsStartChunk{chunk} + summary := rpc.GroupsStartEnd{Started: []string{"api"}, Errors: []string{}} + timedOut, err := waitForStartReady(context.Background(), rpc.GroupsStartParams{ConfigPath: &path}, + nil, chunks, &summary, 0, false) + if err != nil { + t.Fatal(err) + } + if chunks[0].Ready != nil && *chunks[0].Ready { + t.Fatalf("run %s was reported ready on run %s's evidence", "run-r1", replacement) + } + if chunks[0].State != "failed" || chunks[0].Reason != "run_replaced" { + t.Fatalf("chunk = %+v, want a run_replaced failure naming the replacement", chunks[0]) + } + if len(summary.Errors) != 1 || summary.Errors[0] != "api" { + t.Fatalf("summary errors = %v, want api", summary.Errors) + } + if len(timedOut) != 0 { + t.Fatalf("cleanup = %v, want none: the replacement is not this invocation's to stop", timedOut) + } + }) + } +} + +// TestWaitForStartReadyConfirmsItsOwnRun is the other half of the identity +// rule: the same row shape, naming the run this invocation started, is ready. +// A rule that fails every named run would satisfy the test above and break +// every ordinary wait. +func TestWaitForStartReadyConfirmsItsOwnRun(t *testing.T) { + port, healthz := 19001, "/healthz" + path := startManifest(t, " - name: api\n cmd: echo api\n") + started := "run-r1" + startRowsFrom(t, state.Group{Name: "demo", ConfigPath: &path, Services: []state.Service{{ + Name: "api", Running: true, RunID: &started, PortActual: &port, Health: &healthz, + HealthStatus: &state.Health{Status: state.HealthOK}, + }}}) + + chunks := []rpc.GroupsStartChunk{{Service: "api", State: "started", RunID: started, ReadyTimeout: "250ms"}} + summary := rpc.GroupsStartEnd{Started: []string{"api"}, Errors: []string{}} + if _, err := waitForStartReady(context.Background(), rpc.GroupsStartParams{ConfigPath: &path}, + nil, chunks, &summary, 0, false); err != nil { + t.Fatal(err) + } + if chunks[0].Ready == nil || !*chunks[0].Ready { + t.Fatalf("chunk = %+v, want the run this invocation started reported ready", chunks[0]) + } + if len(summary.Errors) != 0 { + t.Fatalf("summary errors = %v, want none", summary.Errors) + } +} + +// TestWaitForStartReadyWillNotReadAnUnattributedListener: the row says +// something is listening and healthy, but it names no run. A listener the row +// cannot attribute is some process on the service's address, not the run this +// wait started, so it does not end the wait as ready. +func TestWaitForStartReadyWillNotReadAnUnattributedListener(t *testing.T) { + port, healthz := 19001, "/healthz" + path := startManifest(t, " - name: api\n cmd: echo api\n") + startRowsFrom(t, state.Group{Name: "demo", ConfigPath: &path, Services: []state.Service{{ + Name: "api", Running: true, PortActual: &port, Health: &healthz, + HealthStatus: &state.Health{Status: state.HealthOK}, + }}}) + + chunks := []rpc.GroupsStartChunk{{Service: "api", State: "started", RunID: "run-r1", ReadyTimeout: "100ms"}} + summary := rpc.GroupsStartEnd{Started: []string{"api"}, Errors: []string{}} + timedOut, err := waitForStartReady(context.Background(), rpc.GroupsStartParams{ConfigPath: &path}, + nil, chunks, &summary, 0, false) + if err != nil { + t.Fatal(err) + } + if chunks[0].Ready != nil && *chunks[0].Ready { + t.Fatal("a listener the row cannot attribute was credited to the run this wait started") + } + if chunks[0].State != "failed" || chunks[0].Reason != "ready_timeout" { + t.Fatalf("chunk = %+v, want the budget to expire unconfirmed", chunks[0]) + } + if len(timedOut) != 1 || timedOut[0].service != "api" || timedOut[0].runID != "run-r1" { + t.Fatalf("cleanup = %v, want the run this invocation started", timedOut) + } +} + +// TestWaitForStartReadyKeepsLegacyChunksUsable: a daemon that predates run ids +// names no run to compare against. There is no identity evidence, so the wait +// keeps reading the service row rather than refusing every old daemon. +func TestWaitForStartReadyKeepsLegacyChunksUsable(t *testing.T) { + port, healthz := 19001, "/healthz" + path := startManifest(t, " - name: api\n cmd: echo api\n") + live := "run-whatever" + startRowsFrom(t, state.Group{Name: "demo", ConfigPath: &path, Services: []state.Service{{ + Name: "api", Running: true, RunID: &live, PortActual: &port, Health: &healthz, + HealthStatus: &state.Health{Status: state.HealthOK}, + }}}) + + chunks := []rpc.GroupsStartChunk{{Service: "api", State: "started", ReadyTimeout: "250ms"}} + summary := rpc.GroupsStartEnd{Started: []string{"api"}, Errors: []string{}} + if _, err := waitForStartReady(context.Background(), rpc.GroupsStartParams{ConfigPath: &path}, + nil, chunks, &summary, 0, false); err != nil { + t.Fatal(err) + } + if chunks[0].Ready == nil || !*chunks[0].Ready { + t.Fatalf("chunk = %+v, want an unnamed run read from the service row", chunks[0]) + } +}