diff --git a/docs/agent/architecture.md b/docs/agent/architecture.md index e4a48e77..6b98dd41 100644 --- a/docs/agent/architecture.md +++ b/docs/agent/architecture.md @@ -120,6 +120,13 @@ A start accepts only readiness evidence for the current service. The no-port run the stop receipt are fixed before the first signal and are not re-derived from the same PID in a later stage. Operations are refused when reliable control identity is missing. +A readiness timeout's cleanup is one of those operations: it names the run this invocation started for +each service (`expect_runs` on `groups.kill`), and the daemon selects only listeners and no-port runs +whose run id matches. A service another controller restarted during the wait is a different run and is +left alone; a target whose identity cannot be verified is not selected either. The CLI refuses to send +a name-only cleanup to a daemon that does not announce the capability, because that daemon would stop +whatever answers to the name. + `down` captures reservations before signalling, reuses a fresh scan after the actual stop, checks the original registry (including compatibility stopping records), and deletes only after comparing the full reservation set inside a short transaction. Failure, cancellation, partial stops, remaining runs or diff --git a/docs/cli.md b/docs/cli.md index 6418b768..292f3cd9 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -477,8 +477,12 @@ it exits 2 without running the start; a service that times out has state `failed `ready_timeout`, the message says whether the timeout came from that service's readiness budget or from the caller's `--wait-timeout` ceiling, and the command exits with code 1. Only runs newly started this time that time out enter stop cleanup; existing runs are not stopped or restarted because this -wait failed. 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 +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` / +`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 `status`, the desktop service row, attention and the event stream do not show it as an ordinary manual stop. Health facts carry `status`, `code`, `reason` and `observed_at`; when a dependency declared diff --git a/docs/cli.zh.md b/docs/cli.zh.md index c18f17c0..ef4e68cd 100644 --- a/docs/cli.zh.md +++ b/docs/cli.zh.md @@ -397,7 +397,10 @@ cwd 中执行，按 `cmd:` 的规则拆成 argv（不经过 shell），支持 `$ 启用 `--wait` 或 `--ready`，否则退出 2，不执行启动； 超时的服务状态为 `failed`、原因是 `ready_timeout`，消息说明超时来自该服务的就绪预算，还是调用方给的 `--wait-timeout` 上限，命令退出码为 1。只有本次新启动的超时运行才进入停止清理； -已经存在的运行不会因这次等待失败而被停止或重启。`started` / `skipped` 汇总描述启动动作， +已经存在的运行不会因这次等待失败而被停止或重启。清理按 `run_id` 只停止本次启动的那个运行： +等待期间另一个控制者为同一服务启动的新运行是另一个运行，不会被停掉。daemon 若未声明 +`groups.kill.expect_runs`，就无法把停止限定到指定运行，因此不会收到只带服务名的清理： +命令失败，并说明超时的运行仍在运行。`started` / `skipped` 汇总描述启动动作， 就绪判定失败另外出现在 `errors` 与逐服务 `state` 中。对本次新启动且被清理的运行， 引擎把同一个 `ready_timeout` 写入服务的最后退出事实，因此 `status`、桌面服务行、attention 和事件流 不会把它显示成普通的手动停止。健康事实带有 `status`、`code`、`reason` 与 `observed_at`， diff --git a/engine/docs/schema/protocol.schema.json b/engine/docs/schema/protocol.schema.json index 65dd25ec..1baa04b0 100644 --- a/engine/docs/schema/protocol.schema.json +++ b/engine/docs/schema/protocol.schema.json @@ -842,6 +842,12 @@ }, "type": "array" }, + "expect_runs": { + "items": { + "$ref": "#/definitions/RunExpectation" + }, + "type": "array" + }, "release": { "type": "boolean" }, @@ -1683,6 +1689,21 @@ "root_pid" ] }, + "RunExpectation": { + "properties": { + "service": { + "type": "string" + }, + "run_id": { + "type": "string" + } + }, + "type": "object", + "required": [ + "service", + "run_id" + ] + }, "RunRecord": { "properties": { "id": { diff --git a/engine/internal/cmd/ready_timeout_test.go b/engine/internal/cmd/ready_timeout_test.go index 1e922ee9..01cefed8 100644 --- a/engine/internal/cmd/ready_timeout_test.go +++ b/engine/internal/cmd/ready_timeout_test.go @@ -94,8 +94,8 @@ func TestWaitForStartReadyUsesEachServiceBudget(t *testing.T) { chdir(t, dir) chunks := []rpc.GroupsStartChunk{ - {Service: "api", State: "started", ReadyTimeout: "15ms"}, - {Service: "worker", State: "started", ReadyTimeout: "35ms"}, + {Service: "api", State: "started", ReadyTimeout: "15ms", RunID: "run-api"}, + {Service: "worker", State: "started", ReadyTimeout: "35ms", RunID: "run-worker"}, } summary := rpc.GroupsStartEnd{Started: []string{"api", "worker"}} timedOut, err := waitForStartReady(context.Background(), @@ -103,9 +103,13 @@ func TestWaitForStartReadyUsesEachServiceBudget(t *testing.T) { if err != nil { t.Fatal(err) } - if len(timedOut) != 2 || timedOut[0] != "api" || timedOut[1] != "worker" { + if len(timedOut) != 2 || timedOut[0].service != "api" || timedOut[1].service != "worker" { t.Fatalf("timed out = %v, want both owned services", timedOut) } + // The cleanup is authorized by run identity, not by the service name. + if timedOut[0].runID != "run-api" || timedOut[1].runID != "run-worker" { + t.Fatalf("timed out = %v, want each service's own run", timedOut) + } byName := map[string]string{} for _, chunk := range chunks { byName[chunk.Service] = chunk.Error diff --git a/engine/internal/cmd/timeout_cleanup_test.go b/engine/internal/cmd/timeout_cleanup_test.go new file mode 100644 index 00000000..919693cd --- /dev/null +++ b/engine/internal/cmd/timeout_cleanup_test.go @@ -0,0 +1,100 @@ +package cmd + +import ( + "context" + "strings" + "testing" + + "github.com/sheathedsharp/option-berth/engine/internal/daemon/rpc" +) + +// fakeKillCaller records the one call a timeout cleanup is allowed to make. +type fakeKillCaller struct { + hello rpc.DaemonHelloResult + calls []struct { + method string + params any + } +} + +func (f *fakeKillCaller) Call(_ context.Context, method string, params, result any) error { + f.calls = append(f.calls, struct { + method string + params any + }{method: method, params: params}) + if env, ok := result.(*rpc.KillEnvelope); ok { + env.MutationResult.OK = true + } + return nil +} + +func (f *fakeKillCaller) Hello() rpc.DaemonHelloResult { return f.hello } + +// The cleanup stops the runs this invocation started, not whatever now answers +// to the service's name; that is the only thing that makes another controller's +// replacement run safe from this timeout. +func TestStopTimedOutServicesNamesTheRunsItStarted(t *testing.T) { + caller := &fakeKillCaller{hello: rpc.DaemonHelloResult{ + Capabilities: []string{rpc.CapabilityKillExpectRuns}, + }} + timedOut := []startTimeout{{service: "api", runID: "run-api"}, {service: "worker", runID: "run-worker"}} + + if err := stopTimedOutServices(context.Background(), caller, rpc.GroupsStartParams{}, timedOut); err != nil { + t.Fatalf("stopTimedOutServices: %v", err) + } + if len(caller.calls) != 1 || caller.calls[0].method != "groups.kill" { + t.Fatalf("calls = %+v, want one groups.kill", caller.calls) + } + params, ok := caller.calls[0].params.(rpc.GroupsKillParams) + if !ok { + t.Fatalf("params = %T, want rpc.GroupsKillParams", caller.calls[0].params) + } + if len(params.Only) != 0 { + t.Fatalf("only = %v, must be empty: a name-scoped stop can reach a replacement run", params.Only) + } + if len(params.ExpectRuns) != 2 || + params.ExpectRuns[0] != (rpc.RunExpectation{Service: "api", RunID: "run-api"}) || + params.ExpectRuns[1] != (rpc.RunExpectation{Service: "worker", RunID: "run-worker"}) { + t.Fatalf("expect_runs = %+v, want both started runs", params.ExpectRuns) + } + if !params.Release || params.Reason != "ready_timeout" { + t.Fatalf("params = %+v, want the readiness-timeout stop", params) + } +} + +// A daemon that predates expect_runs ignores it and stops by name. Sending the +// cleanup anyway is how it reaches the run somebody else started, so the stop +// is refused instead — and the error says the timed-out runs are still up. +func TestStopTimedOutServicesRefusesADaemonThatStopsByName(t *testing.T) { + caller := &fakeKillCaller{hello: rpc.DaemonHelloResult{DaemonVersion: "0.9.0"}} + + err := stopTimedOutServices(context.Background(), caller, + rpc.GroupsStartParams{}, []startTimeout{{service: "api", runID: "run-api"}}) + if err == nil { + t.Fatal("a daemon without expect_runs must not receive a name-only cleanup") + } + if !strings.Contains(err.Error(), "api") || !strings.Contains(err.Error(), "not stopped") || + !strings.Contains(err.Error(), "daemon restart") { + t.Fatalf("error = %v, want the services, what was not done, and the way out", err) + } + if len(caller.calls) != 0 { + t.Fatalf("calls = %+v, want no stop at all", caller.calls) + } +} + +// Same rule for a run this invocation cannot name: without the id there is no +// way to tell the timed-out run from its replacement, so nothing is signalled. +func TestStopTimedOutServicesRefusesAnUnnamedRun(t *testing.T) { + caller := &fakeKillCaller{hello: rpc.DaemonHelloResult{ + Capabilities: []string{rpc.CapabilityKillExpectRuns}, + }} + + err := stopTimedOutServices(context.Background(), caller, + rpc.GroupsStartParams{}, []startTimeout{{service: "api"}}) + if err == nil || !strings.Contains(err.Error(), "api") { + t.Fatalf("error = %v, want a refusal naming the service", err) + } + if len(caller.calls) != 0 { + t.Fatalf("calls = %+v, want no stop at all", caller.calls) + } +} diff --git a/engine/internal/cmd/up.go b/engine/internal/cmd/up.go index b0d5013b..5477f02b 100644 --- a/engine/internal/cmd/up.go +++ b/engine/internal/cmd/up.go @@ -263,15 +263,46 @@ func consumeStartWithWait(ctx context.Context, c *client.Client, stream *client. return nil } +// killCaller is the part of the daemon client a timeout cleanup needs: one +// call and the capability list. It is an interface so the stop request can be +// tested without a live daemon. +type killCaller interface { + Call(ctx context.Context, method string, params, result any) error + Hello() rpc.DaemonHelloResult +} + // stopTimedOutServices makes a readiness timeout terminal. The same group // mutation path used by `oberth down` stops portless workers as well as -// listeners, while Only keeps already-running or skipped services untouched. -func stopTimedOutServices(ctx context.Context, c *client.Client, params rpc.GroupsStartParams, services []string) error { +// listeners, but the stop names the run this invocation started for each +// service. A service name alone would also match a run another controller +// started while this wait was running — that run is not ours to stop and must +// survive the timeout. `expect_runs` also keeps the stop scoped, so +// already-running or skipped services and their `port: auto` claims stay. +// +// A daemon that does not know `expect_runs` would silently drop the field and +// stop by name, which is exactly the wrong run; the cleanup is refused then, +// and an invocation that cannot name the run it started refuses too. +func stopTimedOutServices(ctx context.Context, c killCaller, params rpc.GroupsStartParams, timedOut []startTimeout) error { + if !hasCapability(c.Hello(), rpc.CapabilityKillExpectRuns) { + return timeoutCleanupRefused(c.Hello(), timedOut, + "does not announce "+rpc.CapabilityKillExpectRuns+", so it would stop whatever answers to the service name") + } + expect := make([]rpc.RunExpectation, 0, len(timedOut)) + for _, timedOutService := range timedOut { + if timedOutService.runID == "" { + return timeoutCleanupRefused(c.Hello(), timedOut, + "did not name the run it started for "+timedOutService.service) + } + expect = append(expect, rpc.RunExpectation{ + Service: timedOutService.service, + RunID: timedOutService.runID, + }) + } var env rpc.KillEnvelope err := c.Call(ctx, "groups.kill", rpc.GroupsKillParams{ Name: valueOrEmpty(params.Name), ConfigPath: params.ConfigPath, - Only: services, + ExpectRuns: expect, Release: true, Reason: "ready_timeout", }, &env) @@ -281,6 +312,22 @@ func stopTimedOutServices(ctx context.Context, c *client.Client, params rpc.Grou return nil } +// timeoutCleanupRefused says plainly that the timed-out runs were left running +// and why: a cleanup that stops the wrong run must not read as success, and an +// agent needs the state it is actually in. +func timeoutCleanupRefused(hello rpc.DaemonHelloResult, timedOut []startTimeout, why string) error { + version := hello.DaemonVersion + if version == "" { + version = "an older version" + } + services := make([]string, 0, len(timedOut)) + for _, service := range timedOut { + services = append(services, service.service) + } + return fmt.Errorf("the timed-out runs for %s were not stopped: the running daemon (%s) %s\nhint: they may still be running; stop them with `oberth down`, or restart the daemon with `oberth daemon restart`", + strings.Join(services, ", "), version, why) +} + func valueOrEmpty(value *string) string { if value == nil { return "" diff --git a/engine/internal/cmd/up_ready.go b/engine/internal/cmd/up_ready.go index b659b4bf..70177569 100644 --- a/engine/internal/cmd/up_ready.go +++ b/engine/internal/cmd/up_ready.go @@ -14,6 +14,15 @@ import ( "github.com/sheathedsharp/option-berth/engine/internal/state" ) +// 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 +// started while this wait was still running. +type startTimeout struct { + service string + runID string +} + // waitForStartReady waits on the same state that status exposes. A service // with a configured health path is ready only after its listener reports ok; // every other service is ready once its run is live (or its listener is up). @@ -21,7 +30,7 @@ import ( // second readiness protocol to the daemon. func waitForStartReady(ctx context.Context, params rpc.GroupsStartParams, cfg *groups.Config, chunks []rpc.GroupsStartChunk, summary *rpc.GroupsStartEnd, timeout time.Duration, - timeoutSet bool) ([]string, error) { + timeoutSet bool) ([]startTimeout, error) { waitStart := time.Now() // An explicit --wait-timeout is the ceiling for the whole invocation. It // does not replace each service's own readiness budget; it only caps the @@ -51,7 +60,7 @@ func waitForStartReady(ctx context.Context, params rpc.GroupsStartParams, cfg *g // start would read an exit from this same second as older than the wait. // Align the timestamp fallback to the second; the run id stays primary. fallbackSince := waitStart.UTC().Truncate(time.Second) - var timedOut []string + var timedOut []startTimeout for { pp, gg, obs, err := groupRowsObserved(ctx) if obs != nil && obs.State != string(ports.ObservationObservable) { @@ -94,8 +103,12 @@ func waitForStartReady(ctx context.Context, params rpc.GroupsStartParams, cfg *g } // An older daemon may omit run_id on a skipped chunk. Once // the service is live, the group row names its run, and - // that run is what this wait is about. - if selection.runID == "" && svc.Running && svc.RunID != nil { + // that run is what this wait is about. Only a run this + // invocation did not start is read from the live row: for + // its own start the chunk is the authority, and a run id + // copied from whatever is live now could be a replacement + // that the timeout cleanup must not touch. + if selection.runID == "" && !selection.owned && svc.Running && svc.RunID != nil { selection.runID = *svc.RunID selected[name] = selection } @@ -117,7 +130,7 @@ func waitForStartReady(ctx context.Context, params rpc.GroupsStartParams, cfg *g } } if len(selected) == 0 { - sort.Strings(timedOut) + sortStartTimeouts(timedOut) return timedOut, nil } now := time.Now() @@ -129,12 +142,12 @@ func waitForStartReady(ctx context.Context, params rpc.GroupsStartParams, cfg *g startTimeoutMessage(selection, name, timeout), nil) // Waiting on an existing run does not authorize stopping it. if selection.owned { - timedOut = append(timedOut, name) + timedOut = append(timedOut, startTimeout{service: name, runID: selection.runID}) } delete(selected, name) } if len(selected) == 0 { - sort.Strings(timedOut) + sortStartTimeouts(timedOut) return timedOut, nil } select { @@ -195,6 +208,12 @@ func startChunkReadyTimeout(c rpc.GroupsStartChunk) time.Duration { return groups.DefaultReadyTimeout } +// sortStartTimeouts orders the cleanup list by service, so the stop request +// and its failure message read the same way on every run. +func sortStartTimeouts(timeouts []startTimeout) { + sort.Slice(timeouts, func(i, j int) bool { return timeouts[i].service < timeouts[j].service }) +} + // startSelection is what this invocation selected for one service: the run it // is about, whether this invocation started that run (and may therefore stop // it on timeout), the readiness budget its manifest gives it, and the deadline diff --git a/engine/internal/cmd/up_test.go b/engine/internal/cmd/up_test.go index 8bcd2fd7..79a181b6 100644 --- a/engine/internal/cmd/up_test.go +++ b/engine/internal/cmd/up_test.go @@ -166,7 +166,7 @@ func TestWaitForStartReadyUsesTimeoutAsStructuredFailure(t *testing.T) { summary := rpc.GroupsStartEnd{Started: []string{"api"}, Errors: []string{}} if timedOut, err := waitForStartReady(context.Background(), params, nil, chunks, &summary, time.Millisecond, true); err != nil { t.Fatal(err) - } else if len(timedOut) != 1 || timedOut[0] != "api" { + } else if len(timedOut) != 1 || timedOut[0].service != "api" { t.Fatalf("timed out = %v, want api", timedOut) } if chunks[0].State != "failed" || chunks[0].Reason != "ready_timeout" || @@ -415,14 +415,14 @@ func TestWaitForStartReadyMixedTimeoutCleansOnlyNewStarts(t *testing.T) { t.Fatal(err) } chdir(t, dir) - chunks := []rpc.GroupsStartChunk{{Service: "old", State: "skipped", Skipped: true}, {Service: "new", State: "started"}} + chunks := []rpc.GroupsStartChunk{{Service: "old", State: "skipped", Skipped: true}, {Service: "new", State: "started", RunID: "run-new"}} summary := rpc.GroupsStartEnd{Skipped: []string{"old"}, Started: []string{"new"}} cleanup, err := waitForStartReady(context.Background(), rpc.GroupsStartParams{ConfigPath: &path}, nil, chunks, &summary, time.Millisecond, true) if err != nil { t.Fatal(err) } - if len(cleanup) != 1 || cleanup[0] != "new" { - t.Fatalf("cleanup=%v, must exclude existing service", cleanup) + if len(cleanup) != 1 || cleanup[0].service != "new" || cleanup[0].runID != "run-new" { + t.Fatalf("cleanup=%v, must be the run this invocation started and only it", cleanup) } if len(summary.Errors) != 2 { t.Fatalf("both services must fail readiness: %+v", summary) diff --git a/engine/internal/daemon/group_stop_capture.go b/engine/internal/daemon/group_stop_capture.go index 346cef5f..64aada3f 100644 --- a/engine/internal/daemon/group_stop_capture.go +++ b/engine/internal/daemon/group_stop_capture.go @@ -24,15 +24,18 @@ type groupStopCapturer interface { CaptureGroupStop(group, reason string) GroupStopCapture } -func captureSilentGroupStop(rt *Runtime, snap state.Snapshot, group string, only map[string]bool, reason string, dryRun bool) ([]killer.Target, func(int), error) { +// captureSilentGroupStop selects the runs that hold no port. only selects by +// service name and expect by run identity; the caller never passes both, and +// whichever is set narrows the list before any identity is captured. +func captureSilentGroupStop(rt *Runtime, snap state.Snapshot, group string, only map[string]bool, expect map[string]string, reason string, dryRun bool) ([]killer.Target, func(int), error) { reg := rt.Runs() capturer, ok := reg.(groupStopCapturer) if !ok { // Preserve read-only compatibility. A legacy registry may list PIDs, but // separate GroupRuns/RunStart calls cannot pin their generation atomically. pids := reg.GroupPIDs(group) - if len(only) > 0 { - pids = groupPIDs(reg.GroupRuns(group), only) + if len(only) > 0 || len(expect) > 0 { + pids = groupPIDs(reg.GroupRuns(group), only, expect) } pids = runsWithoutPorts(snap, group, pids) if len(pids) > 0 && !dryRun { @@ -51,6 +54,9 @@ func captureSilentGroupStop(rt *Runtime, snap state.Snapshot, group string, only if len(only) > 0 && !only[run.Owner.Name] { continue } + if len(expect) > 0 && expect[run.Owner.Name] != run.Owner.ID { + continue + } if run.Owner.RootPID <= 0 { return nil, nil, errors.New("invalid captured run PID") } diff --git a/engine/internal/daemon/group_stop_capture_test.go b/engine/internal/daemon/group_stop_capture_test.go index 4719af9f..b3312a6d 100644 --- a/engine/internal/daemon/group_stop_capture_test.go +++ b/engine/internal/daemon/group_stop_capture_test.go @@ -28,7 +28,7 @@ func TestSilentGroupStopUsesCapturedBirthAndScope(t *testing.T) { rt := &Runtime{} rt.SetRuns(reg) snap := state.Snapshot{Ports: []state.Port{{PID: 424241, Port: 12501, Group: ptr("demo")}}} - targets, receipt, err := captureSilentGroupStop(rt, snap, "demo", map[string]bool{"web": true, "worker": true}, "stopped", false) + targets, receipt, err := captureSilentGroupStop(rt, snap, "demo", map[string]bool{"web": true, "worker": true}, nil, "stopped", false) if err != nil || len(targets) != 1 || targets[0].PID != 424242 || targets[0].Name != "worker" || !targets[0].StartedAt.Equal(at) || receipt == nil { t.Fatalf("targets=%+v error=%v", targets, err) } @@ -36,16 +36,37 @@ func TestSilentGroupStopUsesCapturedBirthAndScope(t *testing.T) { if signaled != 424242 { t.Fatal("captured receipt was not used") } - _, receipt, err = captureSilentGroupStop(rt, snap, "demo", nil, "stopped", true) + _, receipt, err = captureSilentGroupStop(rt, snap, "demo", nil, nil, "stopped", true) if err != nil || receipt != nil { t.Fatalf("dry run receipt=%v error=%v", receipt != nil, err) } } +// A timeout cleanup names the run it started. When that service was restarted +// in the meantime, the replacement run answers to the same service name and +// holds no port — it must not be captured for the stop. +func TestSilentGroupStopFollowsTheExpectedRunNotTheService(t *testing.T) { + at := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + reg := &capturedStopFixture{capture: GroupStopCapture{Runs: []GroupStopRun{ + {Owner: state.Run{ID: "run-timed-out", Group: "demo", Name: "worker", RootPID: 424242}, StartedAt: at}, + {Owner: state.Run{ID: "run-replacement", Group: "demo", Name: "worker", RootPID: 424243}, StartedAt: at}, + }}} + rt := &Runtime{} + rt.SetRuns(reg) + targets, _, err := captureSilentGroupStop(rt, state.Snapshot{}, "demo", nil, + map[string]string{"worker": "run-timed-out"}, "", false) + if err != nil { + t.Fatal(err) + } + if len(targets) != 1 || targets[0].PID != 424242 { + t.Fatalf("targets = %+v, want only the run the caller started", targets) + } +} + func TestSilentGroupStopRejectsMissingIdentityBeforeSignals(t *testing.T) { rt := &Runtime{} rt.SetRuns(&capturedStopFixture{capture: GroupStopCapture{Runs: []GroupStopRun{{Owner: state.Run{RootPID: 424242}}}}}) - targets, receipt, err := captureSilentGroupStop(rt, state.Snapshot{}, "demo", nil, "", false) + targets, receipt, err := captureSilentGroupStop(rt, state.Snapshot{}, "demo", nil, nil, "", false) if err == nil || len(targets) != 0 || receipt != nil { t.Fatalf("targets=%+v error=%v", targets, err) } diff --git a/engine/internal/daemon/groups_kill.go b/engine/internal/daemon/groups_kill.go index a023e33c..8d90c9cb 100644 --- a/engine/internal/daemon/groups_kill.go +++ b/engine/internal/daemon/groups_kill.go @@ -29,6 +29,10 @@ func handleGroupsKill(ctx context.Context, req *Request) (any, error) { return nil, rpc.NewError(rpc.CodeInvalidParams, "unsupported stop reason "+p.Reason, "use stopped or ready_timeout") } + if len(p.Only) > 0 && len(p.ExpectRuns) > 0 { + return nil, rpc.NewError(rpc.CodeInvalidParams, "only and expect_runs are mutually exclusive", + "expect_runs names the runs to stop; only names services, whatever run each is") + } target, err := killGroupTarget(req.Runtime, p) if err != nil { return nil, err @@ -46,7 +50,13 @@ func handleGroupsKill(ctx context.Context, req *Request) (any, error) { if err != nil { return nil, err } - scoped := len(only) > 0 + expect, err := expectFilter(req.Runtime, name, cfg, p.ExpectRuns) + if err != nil { + return nil, err + } + // Either selector makes the stop scoped, which is what keeps the group's + // `port: auto` claims attached to the services that keep running. + scoped := len(only) > 0 || len(expect) > 0 var portRelease *servicePortRelease if p.Release && !p.DryRun && !scoped { @@ -72,7 +82,12 @@ func handleGroupsKill(ctx context.Context, req *Request) (any, error) { return nil, err } - targets := groupTargetsFor(snap, name, only) + var targets []killer.Target + if len(expect) > 0 { + targets = expectedGroupTargets(snap, name, expect) + } else { + targets = groupTargetsFor(snap, name, only) + } var pidTargets []killer.Target var runSignal func(int) if p.Release { @@ -83,7 +98,7 @@ func handleGroupsKill(ctx context.Context, req *Request) (any, error) { if cfg == nil { cfg = configForGroup(req.Runtime, name) } - pidTargets, runSignal, err = captureSilentGroupStop(req.Runtime, snap, name, only, p.Reason, p.DryRun) + pidTargets, runSignal, err = captureSilentGroupStop(req.Runtime, snap, name, only, expect, p.Reason, p.DryRun) if err != nil { return nil, rpc.NewError(rpc.CodeInternal, err.Error(), "inspect status before retrying down") } @@ -164,6 +179,65 @@ func serviceFilter(rt *Runtime, name string, cfg *groups.Config, requested []str return selected, nil } +// expectFilter validates the run identities a stop is allowed to touch and +// returns service → run id. Both halves are required: a service alone would +// stop a replacement run that another controller started, which is the race +// this selector exists to close. The service is still checked against the +// manifest so a typo is a caller error, not a quiet no-op. +func expectFilter(rt *Runtime, name string, cfg *groups.Config, requested []rpc.RunExpectation) (map[string]string, error) { + if len(requested) == 0 { + return nil, nil + } + if cfg == nil { + cfg = configForGroup(rt, name) + } + if cfg == nil { + return nil, rpc.NewError(rpc.CodeNotFound, "no manifest describes group "+name, + "run `oberth status` to inspect this worktree") + } + known := make(map[string]bool, len(cfg.Services)) + for _, svc := range cfg.Services { + known[svc.Name] = true + } + expected := make(map[string]string, len(requested)) + for i := range requested { + service, runID, err := expectedRun(requested[i]) + if err != nil { + return nil, err + } + if !known[service] { + return nil, rpc.NewError(rpc.CodeNotFound, + fmt.Sprintf("service %q is not declared by group %s", service, name), + "run `oberth status` to inspect this worktree") + } + if expected[service] != "" { + return nil, rpc.NewError(rpc.CodeInvalidParams, + fmt.Sprintf("expect_runs names service %q more than once", service), + "send one expected run per service") + } + expected[service] = runID + } + return expected, nil +} + +// expectedRun reads one expectation and refuses an incomplete one. A service +// without a run id is the name-only stop this selector replaces, so it is a +// caller error rather than a silent widening. +func expectedRun(raw rpc.RunExpectation) (string, string, error) { + service := strings.TrimSpace(raw.Service) + runID := strings.TrimSpace(raw.RunID) + if service == "" { + return "", "", rpc.NewError(rpc.CodeInvalidParams, "expect_runs contains an empty service name", + "name the service and the run id the stop is allowed to stop") + } + if runID == "" { + return "", "", rpc.NewError(rpc.CodeInvalidParams, + fmt.Sprintf("expect_runs names service %q without a run id", service), + "a service-only stop can reach a replacement run; name the run the caller started") + } + return service, runID, nil +} + // runsWithoutPorts removes runs whose process already appears in the port // targets. Port targets are killed in the first pass; sending the same pid // through the run-only pass would produce a duplicate result and a second @@ -190,10 +264,16 @@ func runsWithoutPorts(snap state.Snapshot, name string, pids []int) []int { return out } -func groupPIDs(runs []state.Run, only map[string]bool) []int { +func groupPIDs(runs []state.Run, only map[string]bool, expect map[string]string) []int { var out []int for _, run := range runs { - if only[run.Name] && run.RootPID > 0 { + if len(only) > 0 && !only[run.Name] { + continue + } + if len(expect) > 0 && expect[run.Name] != run.ID { + continue + } + if run.RootPID > 0 { out = append(out, run.RootPID) } } diff --git a/engine/internal/daemon/groups_kill_expect_test.go b/engine/internal/daemon/groups_kill_expect_test.go new file mode 100644 index 00000000..ffb1b458 --- /dev/null +++ b/engine/internal/daemon/groups_kill_expect_test.go @@ -0,0 +1,148 @@ +package daemon + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "path/filepath" + "strings" + "testing" + + "github.com/sheathedsharp/option-berth/engine/internal/daemon/rpc" + "github.com/sheathedsharp/option-berth/engine/internal/ports" + "github.com/sheathedsharp/option-berth/engine/internal/scanner" + "github.com/sheathedsharp/option-berth/engine/internal/store" +) + +// killRuntime is a daemon runtime whose only observation is the rows the test +// hands it: no process is scanned, no signal can be sent, and every stop in +// these tests is a dry run. +func killRuntime(t *testing.T, rows []ports.ListeningPort) *Runtime { + t.Helper() + st, err := store.Open(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = st.Close() }) + loop := scanner.New(scanner.Options{ + DaemonVersion: "test", + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + Scan: func(scanner.Include) ([]ports.ListeningPort, error) { return rows, nil }, + }) + return &Runtime{ + Store: st, + Scanner: loop, + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + } +} + +// timeoutCleanup sends the groups.kill a readiness timeout would send, as raw +// params so an unknown field is ignored rather than rejected: that is the only +// way a regression can show what the daemon does when it does not know the +// cleanup's run identity. +func timeoutCleanup(t *testing.T, rt *Runtime, params string) rpc.KillEnvelope { + t.Helper() + out, err := handleGroupsKill(context.Background(), &Request{ + Method: "groups.kill", + Params: json.RawMessage(params), + Runtime: rt, + }) + if err != nil { + t.Fatalf("groups.kill: %v", err) + } + env, ok := out.(rpc.KillEnvelope) + if !ok { + t.Fatalf("result = %T, want a kill envelope", out) + } + return env +} + +// A readiness timeout's cleanup names the run it gave up on. Another controller +// that restarted the same service in the meantime owns a different run, and the +// stop must leave it alone. +func TestTimeoutCleanupStopsOnlyTheRunItStarted(t *testing.T) { + manifest := "name: demo\nservices:\n - name: api\n cmd: sleep 1\n port: 19001\n" + + t.Run("the replacement run is not selected", func(t *testing.T) { + rt := killRuntime(t, []ports.ListeningPort{{ + Port: 19001, BindAddress: "127.0.0.1", PID: 424242, + Group: "demo", RunID: "run-replacement", Tag: "api", + RunGroup: "demo", RunRootPID: 424242, + }}) + rt.Scanner.ObserveConfig(checkoutWithManifest(t, manifest)) + + env := timeoutCleanup(t, rt, + `{"name":"demo","expect_runs":[{"service":"api","run_id":"run-timed-out"}],"release":true,"reason":"ready_timeout","dry_run":true}`) + if len(env.Results) != 0 { + t.Fatalf("cleanup selected the replacement run's listener: %+v", env.Results) + } + }) + + t.Run("the run this caller started is still selected", func(t *testing.T) { + rt := killRuntime(t, []ports.ListeningPort{{ + Port: 19001, BindAddress: "127.0.0.1", PID: 424242, + Group: "demo", RunID: "run-timed-out", Tag: "api", + RunGroup: "demo", RunRootPID: 424242, + }}) + rt.Scanner.ObserveConfig(checkoutWithManifest(t, manifest)) + + env := timeoutCleanup(t, rt, + `{"name":"demo","expect_runs":[{"service":"api","run_id":"run-timed-out"}],"release":true,"reason":"ready_timeout","dry_run":true}`) + if len(env.Results) != 1 || env.Results[0].Port != 19001 { + t.Fatalf("results = %+v, want the timed-out run's listener", env.Results) + } + }) +} + +// expect_runs is an ownership claim, so an incomplete or contradictory one is +// a caller error: silently falling back to a service-name stop is what lets a +// cleanup reach a replacement run. +func TestTimeoutCleanupRejectsAnIncompleteClaim(t *testing.T) { + manifest := "name: demo\nservices:\n - name: api\n cmd: sleep 1\n port: 19001\n" + cases := []struct { + name string + params string + code int + want string + }{ + { + name: "both selectors", + params: `{"name":"demo","only":["api"],"expect_runs":[{"service":"api","run_id":"run-1"}]}`, + code: rpc.CodeInvalidParams, want: "mutually exclusive", + }, + { + name: "no run id", + params: `{"name":"demo","expect_runs":[{"service":"api"}]}`, + code: rpc.CodeInvalidParams, want: "without a run id", + }, + { + name: "undeclared service", + params: `{"name":"demo","expect_runs":[{"service":"nope","run_id":"run-1"}]}`, + code: rpc.CodeNotFound, want: "not declared", + }, + { + name: "same service twice", + params: `{"name":"demo","expect_runs":[{"service":"api","run_id":"run-1"},{"service":"api","run_id":"run-2"}]}`, + code: rpc.CodeInvalidParams, want: "more than once", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + rt := killRuntime(t, nil) + rt.Scanner.ObserveConfig(checkoutWithManifest(t, manifest)) + _, err := handleGroupsKill(context.Background(), &Request{ + Method: "groups.kill", Params: json.RawMessage(tc.params), Runtime: rt, + }) + if err == nil { + t.Fatal("want a refusal") + } + if got := rpcCode(err); got != tc.code { + t.Fatalf("code = %d, want %d (%v)", got, tc.code, err) + } + if !strings.Contains(err.Error(), tc.want) { + t.Fatalf("error = %v, want %q", err, tc.want) + } + }) + } +} diff --git a/engine/internal/daemon/kill_handlers.go b/engine/internal/daemon/kill_handlers.go index 59359716..b06c5299 100644 --- a/engine/internal/daemon/kill_handlers.go +++ b/engine/internal/daemon/kill_handlers.go @@ -19,6 +19,7 @@ import ( func init() { RegisterHandler("ports.kill", handlePortsKill) RegisterHandler("groups.kill", handleGroupsKill) + RegisterCapability(rpc.CapabilityKillExpectRuns) } func handlePortsKill(ctx context.Context, req *Request) (any, error) { @@ -202,6 +203,31 @@ func groupTargetsFor(snap state.Snapshot, name string, only map[string]bool) []k return out } +// expectedGroupTargets is the ownership-scoped port pass. A listener is a +// target only when it belongs to a run the caller named: a service whose live +// run is a different one was restarted by somebody else and is not this +// caller's to stop. A row the snapshot cannot attribute to a run, or whose run +// name contradicts the expectation, is left alone as well — an unverifiable +// identity cannot widen a stop. +func expectedGroupTargets(snap state.Snapshot, name string, expect map[string]string) []killer.Target { + wanted := make(map[string]string, len(expect)) + for service, runID := range expect { + wanted[runID] = service + } + var out []killer.Target + for _, p := range snap.Ports { + if p.Run == nil || !inSnapshotGroup(p, name) { + continue + } + service, ok := wanted[p.Run.ID] + if !ok || (p.Run.Name != "" && p.Run.Name != service) { + continue + } + out = append(out, killer.Target{Port: p.Port, BindAddress: p.BindAddress}) + } + return out +} + func inSnapshotGroup(p state.Port, name string) bool { candidates := []string{} if p.Group != nil { diff --git a/engine/internal/daemon/rpc/methods.go b/engine/internal/daemon/rpc/methods.go index 1c49cbb6..5444d8f1 100644 --- a/engine/internal/daemon/rpc/methods.go +++ b/engine/internal/daemon/rpc/methods.go @@ -96,6 +96,13 @@ const ( // stream, and a new client talking to an old daemon simply gets no // progress. CapabilityStartPhases = "groups.start.phases" + + // CapabilityKillExpectRuns covers `expect_runs` in groups.kill: a stop + // scoped to the runs the caller started, so a service another controller + // restarted in the meantime is left alone. An older daemon ignores the + // field and stops whatever answers to the service's name, which is exactly + // the replacement run that field exists to protect. + CapabilityKillExpectRuns = "groups.kill.expect_runs" ) // Include lists the optional per-subscriber enrichments ("stats", "health"). @@ -343,6 +350,14 @@ type GroupsKillParams struct { // Only narrows a project stop to these declared services. It is used by // `oberth restart --only`; scoped stops keep auto-port claims. Only []string `json:"only,omitempty"` + // ExpectRuns names the run each service must still be for this stop to + // touch it. It is the cleanup a readiness timeout uses: it stops the runs + // the caller started, not whatever now answers to the service's name, so a + // run another controller started during the wait is left alone. Mutually + // exclusive with Only — Only selects by service and would widen the stop + // past the named runs. Also scoped: a stop with expectations keeps the + // group's `port: auto` claims. + ExpectRuns []RunExpectation `json:"expect_runs,omitempty"` // Release is `oberth down`: besides the group's listening ports it stops // every run option-berth started in the group, port or not, and releases the // claims the group's `port: auto` services hold. A group with a config @@ -354,6 +369,17 @@ type GroupsKillParams struct { Reason string `json:"reason,omitempty" jsonschema:"enum=stopped,enum=ready_timeout"` } +// RunExpectation is one service's expected run identity in a stop. Both fields +// are required: without the run id the stop would be a name-only stop, which is +// the thing this selector exists to avoid. +type RunExpectation struct { + // Service is a service the group's manifest declares. + Service string `json:"service"` + // RunID is the run this caller started and may stop. A live run with the + // same service name but another id is not a target. + RunID string `json:"run_id"` +} + type GroupsStartParams struct { Name *string `json:"name,omitempty"` ConfigPath *string `json:"config_path,omitempty"`