feat(app): stop the broker with vev kill --broker
- commit
- 472aae140c9295f8510da7045372e52998cb2853
- author
- brice · · 2026-09-23 08:29 UTC
- parent
- bdf8e57
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
6 files · +159 -9
A internal/app/broker_kill_linux.go +55 -0
| @@ -0,0 +1,55 @@ | |||
| 1 | + | //go:build linux | |
| 2 | + | ||
| 3 | + | package app | |
| 4 | + | ||
| 5 | + | import ( | |
| 6 | + | "context" | |
| 7 | + | "errors" | |
| 8 | + | "fmt" | |
| 9 | + | "net" | |
| 10 | + | "os" | |
| 11 | + | "syscall" | |
| 12 | + | ||
| 13 | + | "github.com/bnema/vev/internal/adapters/brokeripc" | |
| 14 | + | ) | |
| 15 | + | ||
| 16 | + | // runKillBroker signals only the process proven by kernel credentials to own | |
| 17 | + | // the current broker socket. Dialing does not launch a broker or mutate state. | |
| 18 | + | func runKillBroker(ctx context.Context) error { | |
| 19 | + | path := brokeripc.SocketPath(productionBrokerLayout().Runtime) | |
| 20 | + | conn, err := (&net.Dialer{}).DialContext(ctx, "unix", path) | |
| 21 | + | if err != nil { | |
| 22 | + | if errors.Is(err, os.ErrNotExist) || errors.Is(err, syscall.ECONNREFUSED) { | |
| 23 | + | fmt.Println("no broker running") | |
| 24 | + | return nil | |
| 25 | + | } | |
| 26 | + | return fmt.Errorf("vev: connect broker to stop: %w", err) | |
| 27 | + | } | |
| 28 | + | defer func() { _ = conn.Close() }() | |
| 29 | + | unix, ok := conn.(*net.UnixConn) | |
| 30 | + | if !ok { | |
| 31 | + | return errors.New("vev: broker endpoint is not a Unix socket") | |
| 32 | + | } | |
| 33 | + | raw, err := unix.SyscallConn() | |
| 34 | + | if err != nil { | |
| 35 | + | return err | |
| 36 | + | } | |
| 37 | + | var peer *syscall.Ucred | |
| 38 | + | var peerErr error | |
| 39 | + | if err := raw.Control(func(fd uintptr) { | |
| 40 | + | peer, peerErr = syscall.GetsockoptUcred(int(fd), syscall.SOL_SOCKET, syscall.SO_PEERCRED) | |
| 41 | + | }); err != nil { | |
| 42 | + | return err | |
| 43 | + | } | |
| 44 | + | if peerErr != nil { | |
| 45 | + | return peerErr | |
| 46 | + | } | |
| 47 | + | if peer == nil || peer.Uid != uint32(os.Geteuid()) || peer.Pid <= 1 || int(peer.Pid) == os.Getpid() { | |
| 48 | + | return errors.New("vev: broker peer identity is invalid") | |
| 49 | + | } | |
| 50 | + | if err := syscall.Kill(int(peer.Pid), syscall.SIGTERM); err != nil { | |
| 51 | + | return fmt.Errorf("vev: stop broker: %w", err) | |
| 52 | + | } | |
| 53 | + | fmt.Println("stopped broker") | |
| 54 | + | return nil | |
| 55 | + | } | |
| @@ -0,0 +1,55 @@ | |||||
| 1 | + | //go:build linux | |||
| 2 | + | ||||
| 3 | + | package app | |||
| 4 | + | ||||
| 5 | + | import ( | |||
| 6 | + | "context" | |||
| 7 | + | "errors" | |||
| 8 | + | "fmt" | |||
| 9 | + | "net" | |||
| 10 | + | "os" | |||
| 11 | + | "syscall" | |||
| 12 | + | ||||
| 13 | + | "github.com/bnema/vev/internal/adapters/brokeripc" | |||
| 14 | + | ) | |||
| 15 | + | ||||
| 16 | + | // runKillBroker signals only the process proven by kernel credentials to own | |||
| 17 | + | // the current broker socket. Dialing does not launch a broker or mutate state. | |||
| 18 | + | func runKillBroker(ctx context.Context) error { | |||
| 19 | + | path := brokeripc.SocketPath(productionBrokerLayout().Runtime) | |||
| 20 | + | conn, err := (&net.Dialer{}).DialContext(ctx, "unix", path) | |||
| 21 | + | if err != nil { | |||
| 22 | + | if errors.Is(err, os.ErrNotExist) || errors.Is(err, syscall.ECONNREFUSED) { | |||
| 23 | + | fmt.Println("no broker running") | |||
| 24 | + | return nil | |||
| 25 | + | } | |||
| 26 | + | return fmt.Errorf("vev: connect broker to stop: %w", err) | |||
| 27 | + | } | |||
| 28 | + | defer func() { _ = conn.Close() }() | |||
| 29 | + | unix, ok := conn.(*net.UnixConn) | |||
| 30 | + | if !ok { | |||
| 31 | + | return errors.New("vev: broker endpoint is not a Unix socket") | |||
| 32 | + | } | |||
| 33 | + | raw, err := unix.SyscallConn() | |||
| 34 | + | if err != nil { | |||
| 35 | + | return err | |||
| 36 | + | } | |||
| 37 | + | var peer *syscall.Ucred | |||
| 38 | + | var peerErr error | |||
| 39 | + | if err := raw.Control(func(fd uintptr) { | |||
| 40 | + | peer, peerErr = syscall.GetsockoptUcred(int(fd), syscall.SOL_SOCKET, syscall.SO_PEERCRED) | |||
| 41 | + | }); err != nil { | |||
| 42 | + | return err | |||
| 43 | + | } | |||
| 44 | + | if peerErr != nil { | |||
| 45 | + | return peerErr | |||
| 46 | + | } | |||
| 47 | + | if peer == nil || peer.Uid != uint32(os.Geteuid()) || peer.Pid <= 1 || int(peer.Pid) == os.Getpid() { | |||
| 48 | + | return errors.New("vev: broker peer identity is invalid") | |||
| 49 | + | } | |||
| 50 | + | if err := syscall.Kill(int(peer.Pid), syscall.SIGTERM); err != nil { | |||
| 51 | + | return fmt.Errorf("vev: stop broker: %w", err) | |||
| 52 | + | } | |||
| 53 | + | fmt.Println("stopped broker") | |||
| 54 | + | return nil | |||
| 55 | + | } | |||
A internal/app/broker_kill_other.go +12 -0
| @@ -0,0 +1,12 @@ | |||
| 1 | + | //go:build !linux | |
| 2 | + | ||
| 3 | + | package app | |
| 4 | + | ||
| 5 | + | import ( | |
| 6 | + | "context" | |
| 7 | + | "errors" | |
| 8 | + | ) | |
| 9 | + | ||
| 10 | + | func runKillBroker(context.Context) error { | |
| 11 | + | return errors.New("vev: stopping the broker requires Linux peer credentials") | |
| 12 | + | } | |
| @@ -0,0 +1,12 @@ | |||||
| 1 | + | //go:build !linux | |||
| 2 | + | ||||
| 3 | + | package app | |||
| 4 | + | ||||
| 5 | + | import ( | |||
| 6 | + | "context" | |||
| 7 | + | "errors" | |||
| 8 | + | ) | |||
| 9 | + | ||||
| 10 | + | func runKillBroker(context.Context) error { | |||
| 11 | + | return errors.New("vev: stopping the broker requires Linux peer credentials") | |||
| 12 | + | } | |||
M internal/app/broker_production_connect.go +17 -0
| @@ -13,6 +13,23 @@ import ( | |||
| 13 | 13 | ||
| 14 | 14 | const productionBrokerStartupTimeout = 10 * time.Second | |
| 15 | 15 | ||
| 16 | + | // connectExistingBroker dials without ensuring or spawning a broker. | |
| 17 | + | func connectExistingBroker(ctx context.Context) (ports.BrokerService, error) { | |
| 18 | + | path := brokeripc.SocketPath(productionBrokerLayout().Runtime) | |
| 19 | + | service, err := brokeripc.NewConnector(path, brokeripc.Config{}).Connect(ctx) | |
| 20 | + | if err != nil { | |
| 21 | + | if backendAbsent(err) { | |
| 22 | + | return nil, absentError(path) | |
| 23 | + | } | |
| 24 | + | return nil, err | |
| 25 | + | } | |
| 26 | + | if err := awaitBrokerPublication(ctx, service); err != nil { | |
| 27 | + | _ = service.Close() | |
| 28 | + | return nil, err | |
| 29 | + | } | |
| 30 | + | return service, nil | |
| 31 | + | } | |
| 32 | + | ||
| 16 | 33 | // connectProductionBroker connects to the sole per-user production broker. | |
| 17 | 34 | // Process election and idle lifetime remain owned by the existing broker | |
| 18 | 35 | // launcher and broker.Supervisor; callers receive only the semantic façade. | |
| @@ -13,6 +13,23 @@ import ( | |||||
| 13 | 13 | ||||
| 14 | const productionBrokerStartupTimeout = 10 * time.Second | 14 | const productionBrokerStartupTimeout = 10 * time.Second | ||
| 15 | 15 | ||||
| 16 | + | // connectExistingBroker dials without ensuring or spawning a broker. | |||
| 17 | + | func connectExistingBroker(ctx context.Context) (ports.BrokerService, error) { | |||
| 18 | + | path := brokeripc.SocketPath(productionBrokerLayout().Runtime) | |||
| 19 | + | service, err := brokeripc.NewConnector(path, brokeripc.Config{}).Connect(ctx) | |||
| 20 | + | if err != nil { | |||
| 21 | + | if backendAbsent(err) { | |||
| 22 | + | return nil, absentError(path) | |||
| 23 | + | } | |||
| 24 | + | return nil, err | |||
| 25 | + | } | |||
| 26 | + | if err := awaitBrokerPublication(ctx, service); err != nil { | |||
| 27 | + | _ = service.Close() | |||
| 28 | + | return nil, err | |||
| 29 | + | } | |||
| 30 | + | return service, nil | |||
| 31 | + | } | |||
| 32 | + | ||||
| 16 | // connectProductionBroker connects to the sole per-user production broker. | 33 | // connectProductionBroker connects to the sole per-user production broker. | ||
| 17 | // Process election and idle lifetime remain owned by the existing broker | 34 | // Process election and idle lifetime remain owned by the existing broker | ||
| 18 | // launcher and broker.Supervisor; callers receive only the semantic façade. | 35 | // launcher and broker.Supervisor; callers receive only the semantic façade. | ||
M internal/app/run.go +25 -8
| @@ -99,6 +99,7 @@ type command struct { | |||
| 99 | 99 | hostTarget string | |
| 100 | 100 | killAll bool | |
| 101 | 101 | killDaemon bool | |
| 102 | + | killBroker bool | |
| 102 | 103 | cmd cmdInvocation | |
| 103 | 104 | brokerServe brokerServeOptions | |
| 104 | 105 | brokerClient brokerClientOptions | |
| @@ -140,6 +141,7 @@ usage: | |||
| 140 | 141 | vev kill <name> kill a session | |
| 141 | 142 | vev kill --all kill all sessions (the daemon keeps running) | |
| 142 | 143 | vev kill --daemon stop the active vev daemon | |
| 144 | + | vev kill --broker stop the active vev broker | |
| 143 | 145 | vev cmd <command> run a control command (vev cmd --help) | |
| 144 | 146 | vev --ui-observe expose passive observation for this interactive client | |
| 145 | 147 | (optional: --ui-socket PATH) | |
| @@ -346,7 +348,7 @@ parsedUIFlags: | |||
| 346 | 348 | return command{kind: kindCmd, cmd: invocation}, nil | |
| 347 | 349 | case "kill": | |
| 348 | 350 | if len(args) < 2 || args[1] == "" { | |
| 349 | - | return command{}, usagef("`kill` requires a session name, --all, or --daemon") | |
| 351 | + | return command{}, usagef("`kill` requires a session name, --all, --daemon, or --broker") | |
| 350 | 352 | } | |
| 351 | 353 | if args[1] == "--" { | |
| 352 | 354 | if len(args) != 3 || args[2] == "" { | |
| @@ -355,13 +357,15 @@ parsedUIFlags: | |||
| 355 | 357 | return command{kind: kindKill, name: args[2]}, nil | |
| 356 | 358 | } | |
| 357 | 359 | if len(args) > 2 { | |
| 358 | - | return command{}, usagef("`kill` accepts exactly one session name, --all, or --daemon") | |
| 360 | + | return command{}, usagef("`kill` accepts exactly one session name, --all, --daemon, or --broker") | |
| 359 | 361 | } | |
| 360 | 362 | switch args[1] { | |
| 361 | 363 | case "--all": | |
| 362 | 364 | return command{kind: kindKill, killAll: true}, nil | |
| 363 | 365 | case "--daemon": | |
| 364 | 366 | return command{kind: kindKill, killDaemon: true}, nil | |
| 367 | + | case "--broker": | |
| 368 | + | return command{kind: kindKill, killBroker: true}, nil | |
| 365 | 369 | default: | |
| 366 | 370 | return command{kind: kindKill, name: args[1]}, nil | |
| 367 | 371 | } | |
| @@ -428,6 +432,9 @@ func dispatch(ctx context.Context, cmd command) error { | |||
| 428 | 432 | case kindHost: | |
| 429 | 433 | return runHostCommand(ctx, cmd, defaultRemoteHostDeps()) | |
| 430 | 434 | case kindKill: | |
| 435 | + | if cmd.killBroker { | |
| 436 | + | return runKillBroker(ctx) | |
| 437 | + | } | |
| 431 | 438 | return runKill(ctx, cmd.name, cmd.killAll, cmd.killDaemon) | |
| 432 | 439 | case kindCmd: | |
| 433 | 440 | return runCmd(ctx, cmd.cmd) | |
| @@ -491,6 +498,7 @@ func parseHostArgs(args []string) (command, error) { | |||
| 491 | 498 | // Composition-root factory seams keep observer propagation testable without | |
| 492 | 499 | // opening real transports. | |
| 493 | 500 | var connectBroker = connectProductionBroker | |
| 501 | + | var connectDaemonStopBroker = connectExistingBroker | |
| 494 | 502 | ||
| 495 | 503 | var newPerformanceTrace = performanceTrace | |
| 496 | 504 | ||
| @@ -920,7 +928,7 @@ func runAttach(ctx context.Context, intent uint8, name, remoteTarget string) (re | |||
| 920 | 928 | } | |
| 921 | 929 | ctx, stop := signal.NotifyContext(ctx, syscall.SIGHUP, syscall.SIGTERM, syscall.SIGINT) | |
| 922 | 930 | defer stop() | |
| 923 | - | _, logCloser, err := configureLogging(logging.Client, false) | |
| 931 | + | log, logCloser, err := configureLogging(logging.Client, false) | |
| 924 | 932 | if err != nil { | |
| 925 | 933 | return err | |
| 926 | 934 | } | |
| @@ -950,7 +958,7 @@ func runAttach(ctx context.Context, intent uint8, name, remoteTarget string) (re | |||
| 950 | 958 | } | |
| 951 | 959 | callbacks := terminalBrokerCallbacks() | |
| 952 | 960 | return runBrokerClient(ctx, brokerClientConfig{ | |
| 953 | - | Connector: withPreconnected(newProductionBrokerConnector(), preconnected), Terminal: terminal, Clock: clk, | |
| 961 | + | Logger: log, Connector: withPreconnected(newProductionBrokerConnector(), preconnected), Terminal: terminal, Clock: clk, | |
| 954 | 962 | InitialNavigation: navigation, ResolveInitialNavigation: resolver, | |
| 955 | 963 | AttachmentEnvironment: terminalAttachmentEnvironment(), SessionEnvironment: terminalSessionEnvironment(), | |
| 956 | 964 | OnState: callbacks.OnState, OnLifecycle: callbacks.OnLifecycle, OnFailure: callbacks.OnFailure, | |
| @@ -1021,8 +1029,11 @@ func runUIRemoteCleanup(ctx context.Context) error { | |||
| 1021 | 1029 | func requestDaemonStop(ctx context.Context) error { | |
| 1022 | 1030 | ctx, cancel := context.WithTimeout(ctx, daemonStopTimeout) | |
| 1023 | 1031 | defer cancel() | |
| 1024 | - | service, err := connectBroker(ctx) | |
| 1032 | + | service, err := connectDaemonStopBroker(ctx) | |
| 1025 | 1033 | if err != nil { | |
| 1034 | + | if errors.Is(err, errBrokerAbsent) { | |
| 1035 | + | return errDaemonNotRunning | |
| 1036 | + | } | |
| 1026 | 1037 | return unreachableBrokerError(err) | |
| 1027 | 1038 | } | |
| 1028 | 1039 | defer func() { _ = service.Close() }() | |
| @@ -1353,8 +1364,16 @@ func unreadableCatalogueError(stateDir string) error { | |||
| 1353 | 1364 | " rm -rf %s", persist.ErrCatalogueUnreadable, stateDir, stateDir) | |
| 1354 | 1365 | } | |
| 1355 | 1366 | ||
| 1356 | - | // runKill executes all kill operations through the per-user broker. | |
| 1367 | + | // runKill uses the existing broker for daemon shutdown and connect-or-spawn | |
| 1368 | + | // for session mutations. | |
| 1357 | 1369 | func runKill(ctx context.Context, name string, all, daemon bool) error { | |
| 1370 | + | if daemon { | |
| 1371 | + | err := requestDaemonStop(ctx) | |
| 1372 | + | if err == nil { | |
| 1373 | + | printKillSuccess(name, all, daemon) | |
| 1374 | + | } | |
| 1375 | + | return err | |
| 1376 | + | } | |
| 1358 | 1377 | service, err := connectBroker(ctx) | |
| 1359 | 1378 | if err != nil { | |
| 1360 | 1379 | return unreachableBrokerError(err) | |
| @@ -1367,8 +1386,6 @@ func runKill(ctx context.Context, name string, all, daemon bool) error { | |||
| 1367 | 1386 | route := localBrokerOperationRoute(service.Snapshot()) | |
| 1368 | 1387 | var result protocol.KillResult | |
| 1369 | 1388 | switch { | |
| 1370 | - | case daemon: | |
| 1371 | - | result, err = operations.StopDaemon(ctx, route) | |
| 1372 | 1389 | case all: | |
| 1373 | 1390 | result, err = operations.KillAll(ctx, route) | |
| 1374 | 1391 | default: | |
| @@ -99,6 +99,7 @@ type command struct { | |||||
| 99 | hostTarget string | 99 | hostTarget string | ||
| 100 | killAll bool | 100 | killAll bool | ||
| 101 | killDaemon bool | 101 | killDaemon bool | ||
| 102 | + | killBroker bool | |||
| 102 | cmd cmdInvocation | 103 | cmd cmdInvocation | ||
| 103 | brokerServe brokerServeOptions | 104 | brokerServe brokerServeOptions | ||
| 104 | brokerClient brokerClientOptions | 105 | brokerClient brokerClientOptions | ||
| @@ -140,6 +141,7 @@ usage: | |||||
| 140 | vev kill <name> kill a session | 141 | vev kill <name> kill a session | ||
| 141 | vev kill --all kill all sessions (the daemon keeps running) | 142 | vev kill --all kill all sessions (the daemon keeps running) | ||
| 142 | vev kill --daemon stop the active vev daemon | 143 | vev kill --daemon stop the active vev daemon | ||
| 144 | + | vev kill --broker stop the active vev broker | |||
| 143 | vev cmd <command> run a control command (vev cmd --help) | 145 | vev cmd <command> run a control command (vev cmd --help) | ||
| 144 | vev --ui-observe expose passive observation for this interactive client | 146 | vev --ui-observe expose passive observation for this interactive client | ||
| 145 | (optional: --ui-socket PATH) | 147 | (optional: --ui-socket PATH) | ||
| @@ -346,7 +348,7 @@ parsedUIFlags: | |||||
| 346 | return command{kind: kindCmd, cmd: invocation}, nil | 348 | return command{kind: kindCmd, cmd: invocation}, nil | ||
| 347 | case "kill": | 349 | case "kill": | ||
| 348 | if len(args) < 2 || args[1] == "" { | 350 | if len(args) < 2 || args[1] == "" { | ||
| 349 | - | return command{}, usagef("`kill` requires a session name, --all, or --daemon") | 351 | + | return command{}, usagef("`kill` requires a session name, --all, --daemon, or --broker") |
| 350 | } | 352 | } | ||
| 351 | if args[1] == "--" { | 353 | if args[1] == "--" { | ||
| 352 | if len(args) != 3 || args[2] == "" { | 354 | if len(args) != 3 || args[2] == "" { | ||
| @@ -355,13 +357,15 @@ parsedUIFlags: | |||||
| 355 | return command{kind: kindKill, name: args[2]}, nil | 357 | return command{kind: kindKill, name: args[2]}, nil | ||
| 356 | } | 358 | } | ||
| 357 | if len(args) > 2 { | 359 | if len(args) > 2 { | ||
| 358 | - | return command{}, usagef("`kill` accepts exactly one session name, --all, or --daemon") | 360 | + | return command{}, usagef("`kill` accepts exactly one session name, --all, --daemon, or --broker") |
| 359 | } | 361 | } | ||
| 360 | switch args[1] { | 362 | switch args[1] { | ||
| 361 | case "--all": | 363 | case "--all": | ||
| 362 | return command{kind: kindKill, killAll: true}, nil | 364 | return command{kind: kindKill, killAll: true}, nil | ||
| 363 | case "--daemon": | 365 | case "--daemon": | ||
| 364 | return command{kind: kindKill, killDaemon: true}, nil | 366 | return command{kind: kindKill, killDaemon: true}, nil | ||
| 367 | + | case "--broker": | |||
| 368 | + | return command{kind: kindKill, killBroker: true}, nil | |||
| 365 | default: | 369 | default: | ||
| 366 | return command{kind: kindKill, name: args[1]}, nil | 370 | return command{kind: kindKill, name: args[1]}, nil | ||
| 367 | } | 371 | } | ||
| @@ -428,6 +432,9 @@ func dispatch(ctx context.Context, cmd command) error { | |||||
| 428 | case kindHost: | 432 | case kindHost: | ||
| 429 | return runHostCommand(ctx, cmd, defaultRemoteHostDeps()) | 433 | return runHostCommand(ctx, cmd, defaultRemoteHostDeps()) | ||
| 430 | case kindKill: | 434 | case kindKill: | ||
| 435 | + | if cmd.killBroker { | |||
| 436 | + | return runKillBroker(ctx) | |||
| 437 | + | } | |||
| 431 | return runKill(ctx, cmd.name, cmd.killAll, cmd.killDaemon) | 438 | return runKill(ctx, cmd.name, cmd.killAll, cmd.killDaemon) | ||
| 432 | case kindCmd: | 439 | case kindCmd: | ||
| 433 | return runCmd(ctx, cmd.cmd) | 440 | return runCmd(ctx, cmd.cmd) | ||
| @@ -491,6 +498,7 @@ func parseHostArgs(args []string) (command, error) { | |||||
| 491 | // Composition-root factory seams keep observer propagation testable without | 498 | // Composition-root factory seams keep observer propagation testable without | ||
| 492 | // opening real transports. | 499 | // opening real transports. | ||
| 493 | var connectBroker = connectProductionBroker | 500 | var connectBroker = connectProductionBroker | ||
| 501 | + | var connectDaemonStopBroker = connectExistingBroker | |||
| 494 | 502 | ||||
| 495 | var newPerformanceTrace = performanceTrace | 503 | var newPerformanceTrace = performanceTrace | ||
| 496 | 504 | ||||
| @@ -920,7 +928,7 @@ func runAttach(ctx context.Context, intent uint8, name, remoteTarget string) (re | |||||
| 920 | } | 928 | } | ||
| 921 | ctx, stop := signal.NotifyContext(ctx, syscall.SIGHUP, syscall.SIGTERM, syscall.SIGINT) | 929 | ctx, stop := signal.NotifyContext(ctx, syscall.SIGHUP, syscall.SIGTERM, syscall.SIGINT) | ||
| 922 | defer stop() | 930 | defer stop() | ||
| 923 | - | _, logCloser, err := configureLogging(logging.Client, false) | 931 | + | log, logCloser, err := configureLogging(logging.Client, false) |
| 924 | if err != nil { | 932 | if err != nil { | ||
| 925 | return err | 933 | return err | ||
| 926 | } | 934 | } | ||
| @@ -950,7 +958,7 @@ func runAttach(ctx context.Context, intent uint8, name, remoteTarget string) (re | |||||
| 950 | } | 958 | } | ||
| 951 | callbacks := terminalBrokerCallbacks() | 959 | callbacks := terminalBrokerCallbacks() | ||
| 952 | return runBrokerClient(ctx, brokerClientConfig{ | 960 | return runBrokerClient(ctx, brokerClientConfig{ | ||
| 953 | - | Connector: withPreconnected(newProductionBrokerConnector(), preconnected), Terminal: terminal, Clock: clk, | 961 | + | Logger: log, Connector: withPreconnected(newProductionBrokerConnector(), preconnected), Terminal: terminal, Clock: clk, |
| 954 | InitialNavigation: navigation, ResolveInitialNavigation: resolver, | 962 | InitialNavigation: navigation, ResolveInitialNavigation: resolver, | ||
| 955 | AttachmentEnvironment: terminalAttachmentEnvironment(), SessionEnvironment: terminalSessionEnvironment(), | 963 | AttachmentEnvironment: terminalAttachmentEnvironment(), SessionEnvironment: terminalSessionEnvironment(), | ||
| 956 | OnState: callbacks.OnState, OnLifecycle: callbacks.OnLifecycle, OnFailure: callbacks.OnFailure, | 964 | OnState: callbacks.OnState, OnLifecycle: callbacks.OnLifecycle, OnFailure: callbacks.OnFailure, | ||
| @@ -1021,8 +1029,11 @@ func runUIRemoteCleanup(ctx context.Context) error { | |||||
| 1021 | func requestDaemonStop(ctx context.Context) error { | 1029 | func requestDaemonStop(ctx context.Context) error { | ||
| 1022 | ctx, cancel := context.WithTimeout(ctx, daemonStopTimeout) | 1030 | ctx, cancel := context.WithTimeout(ctx, daemonStopTimeout) | ||
| 1023 | defer cancel() | 1031 | defer cancel() | ||
| 1024 | - | service, err := connectBroker(ctx) | 1032 | + | service, err := connectDaemonStopBroker(ctx) |
| 1025 | if err != nil { | 1033 | if err != nil { | ||
| 1034 | + | if errors.Is(err, errBrokerAbsent) { | |||
| 1035 | + | return errDaemonNotRunning | |||
| 1036 | + | } | |||
| 1026 | return unreachableBrokerError(err) | 1037 | return unreachableBrokerError(err) | ||
| 1027 | } | 1038 | } | ||
| 1028 | defer func() { _ = service.Close() }() | 1039 | defer func() { _ = service.Close() }() | ||
| @@ -1353,8 +1364,16 @@ func unreadableCatalogueError(stateDir string) error { | |||||
| 1353 | " rm -rf %s", persist.ErrCatalogueUnreadable, stateDir, stateDir) | 1364 | " rm -rf %s", persist.ErrCatalogueUnreadable, stateDir, stateDir) | ||
| 1354 | } | 1365 | } | ||
| 1355 | 1366 | ||||
| 1356 | - | // runKill executes all kill operations through the per-user broker. | 1367 | + | // runKill uses the existing broker for daemon shutdown and connect-or-spawn |
| 1368 | + | // for session mutations. | |||
| 1357 | func runKill(ctx context.Context, name string, all, daemon bool) error { | 1369 | func runKill(ctx context.Context, name string, all, daemon bool) error { | ||
| 1370 | + | if daemon { | |||
| 1371 | + | err := requestDaemonStop(ctx) | |||
| 1372 | + | if err == nil { | |||
| 1373 | + | printKillSuccess(name, all, daemon) | |||
| 1374 | + | } | |||
| 1375 | + | return err | |||
| 1376 | + | } | |||
| 1358 | service, err := connectBroker(ctx) | 1377 | service, err := connectBroker(ctx) | ||
| 1359 | if err != nil { | 1378 | if err != nil { | ||
| 1360 | return unreachableBrokerError(err) | 1379 | return unreachableBrokerError(err) | ||
| @@ -1367,8 +1386,6 @@ func runKill(ctx context.Context, name string, all, daemon bool) error { | |||||
| 1367 | route := localBrokerOperationRoute(service.Snapshot()) | 1386 | route := localBrokerOperationRoute(service.Snapshot()) | ||
| 1368 | var result protocol.KillResult | 1387 | var result protocol.KillResult | ||
| 1369 | switch { | 1388 | switch { | ||
| 1370 | - | case daemon: | |||
| 1371 | - | result, err = operations.StopDaemon(ctx, route) | |||
| 1372 | case all: | 1389 | case all: | ||
| 1373 | result, err = operations.KillAll(ctx, route) | 1390 | result, err = operations.KillAll(ctx, route) | ||
| 1374 | default: | 1391 | default: | ||
M internal/app/run_broker_test.go +44 -1
| @@ -11,6 +11,7 @@ package app | |||
| 11 | 11 | // is an exit-3 unknown outcome that is never read as success. | |
| 12 | 12 | ||
| 13 | 13 | import ( | |
| 14 | + | "bytes" | |
| 14 | 15 | "context" | |
| 15 | 16 | "errors" | |
| 16 | 17 | "io" | |
| @@ -204,8 +205,13 @@ var _ ports.BrokerService = (*seamBrokerService)(nil) | |||
| 204 | 205 | func withSeamBroker(t *testing.T, service ports.BrokerService) { | |
| 205 | 206 | t.Helper() | |
| 206 | 207 | previous := connectBroker | |
| 208 | + | previousStop := connectDaemonStopBroker | |
| 207 | 209 | connectBroker = func(context.Context) (ports.BrokerService, error) { return service, nil } | |
| 208 | - | t.Cleanup(func() { connectBroker = previous }) | |
| 210 | + | connectDaemonStopBroker = connectBroker | |
| 211 | + | t.Cleanup(func() { | |
| 212 | + | connectBroker = previous | |
| 213 | + | connectDaemonStopBroker = previousStop | |
| 214 | + | }) | |
| 209 | 215 | } | |
| 210 | 216 | ||
| 211 | 217 | func seamKillResult(requestID uint64, outcome protocol.KillOutcome, code uint16, text string) protocol.KillResult { | |
| @@ -215,6 +221,43 @@ func seamKillResult(requestID uint64, outcome protocol.KillOutcome, code uint16, | |||
| 215 | 221 | // TestRunListRendersBrokerSessionsWithoutADaemon pins that runList reads the | |
| 216 | 222 | // live session list through BrokerOperations over the per-user broker and | |
| 217 | 223 | // renders it, so a local list never reads durable state or dials a daemon. | |
| 224 | + | func TestRunBrokerSnapshotListReportsNoDaemon(t *testing.T) { | |
| 225 | + | remote := ports.BrokerDaemonObservation{ | |
| 226 | + | Endpoint: "demo@host.test", DisplayOrigin: "demo@host.test", | |
| 227 | + | Registration: domain.RemoteRegistration{Endpoint: "demo@host.test", Incarnation: [16]byte{1}, Generation: 1}, | |
| 228 | + | Policy: seamBrokerTestPolicy(), Availability: domain.RemoteAvailabilityNoDaemon, | |
| 229 | + | Sessions: []catalogue.RemoteCatalogSession{}, | |
| 230 | + | } | |
| 231 | + | snapshot := ports.BrokerSnapshot{Epoch: seamBrokerTestEpoch, Revision: 1, Daemons: []ports.BrokerDaemonObservation{remote}} | |
| 232 | + | ||
| 233 | + | var all bytes.Buffer | |
| 234 | + | require.NoError(t, runBrokerSnapshotList(command{listAll: true}, snapshot, &all)) | |
| 235 | + | require.Contains(t, all.String(), "demo@host.test: no vev daemon") | |
| 236 | + | require.Contains(t, all.String(), "Enter in the picker to create a session") | |
| 237 | + | require.NotContains(t, all.String(), "no sessions") | |
| 238 | + | ||
| 239 | + | var host bytes.Buffer | |
| 240 | + | require.NoError(t, runBrokerSnapshotList(command{listHost: "demo@host.test"}, snapshot, &host)) | |
| 241 | + | require.Contains(t, host.String(), "no vev daemon") | |
| 242 | + | } | |
| 243 | + | ||
| 244 | + | func TestRunBrokerSnapshotListAllShowsNoDaemonAlongsideSessions(t *testing.T) { | |
| 245 | + | remote := ports.BrokerDaemonObservation{ | |
| 246 | + | Endpoint: "demo@host.test", DisplayOrigin: "demo@host.test", | |
| 247 | + | Registration: domain.RemoteRegistration{Endpoint: "demo@host.test", Incarnation: [16]byte{1}, Generation: 1}, | |
| 248 | + | Policy: seamBrokerTestPolicy(), Availability: domain.RemoteAvailabilityNoDaemon, | |
| 249 | + | Sessions: []catalogue.RemoteCatalogSession{}, | |
| 250 | + | } | |
| 251 | + | local := seamBrokerTestSnapshot().Daemons[0] | |
| 252 | + | local.InventoryKnown = true | |
| 253 | + | local.Sessions = []catalogue.RemoteCatalogSession{{LifecycleID: domain.SessionLifecycleID{1}, Name: "work", State: catalogue.RemoteCatalogSessionUp, Tabs: []catalogue.RemoteCatalogTab{}}} | |
| 254 | + | snapshot := ports.BrokerSnapshot{Epoch: seamBrokerTestEpoch, Revision: 1, Daemons: []ports.BrokerDaemonObservation{local, remote}} | |
| 255 | + | var output bytes.Buffer | |
| 256 | + | require.NoError(t, runBrokerSnapshotList(command{listAll: true}, snapshot, &output)) | |
| 257 | + | require.Contains(t, output.String(), "demo@host.test: no vev daemon") | |
| 258 | + | require.Contains(t, output.String(), "work") | |
| 259 | + | } | |
| 260 | + | ||
| 218 | 261 | func TestRunListRendersBrokerSessionsWithoutADaemon(t *testing.T) { | |
| 219 | 262 | stream := newSeamBrokerStream(protocol.Sessions{Sessions: []protocol.SessionInfo{ | |
| 220 | 263 | {Name: "work", State: protocol.SessionUp, Tabs: 2, Attached: true}, | |
| @@ -11,6 +11,7 @@ package app | |||||
| 11 | // is an exit-3 unknown outcome that is never read as success. | 11 | // is an exit-3 unknown outcome that is never read as success. | ||
| 12 | 12 | ||||
| 13 | import ( | 13 | import ( | ||
| 14 | + | "bytes" | |||
| 14 | "context" | 15 | "context" | ||
| 15 | "errors" | 16 | "errors" | ||
| 16 | "io" | 17 | "io" | ||
| @@ -204,8 +205,13 @@ var _ ports.BrokerService = (*seamBrokerService)(nil) | |||||
| 204 | func withSeamBroker(t *testing.T, service ports.BrokerService) { | 205 | func withSeamBroker(t *testing.T, service ports.BrokerService) { | ||
| 205 | t.Helper() | 206 | t.Helper() | ||
| 206 | previous := connectBroker | 207 | previous := connectBroker | ||
| 208 | + | previousStop := connectDaemonStopBroker | |||
| 207 | connectBroker = func(context.Context) (ports.BrokerService, error) { return service, nil } | 209 | connectBroker = func(context.Context) (ports.BrokerService, error) { return service, nil } | ||
| 208 | - | t.Cleanup(func() { connectBroker = previous }) | 210 | + | connectDaemonStopBroker = connectBroker |
| 211 | + | t.Cleanup(func() { | |||
| 212 | + | connectBroker = previous | |||
| 213 | + | connectDaemonStopBroker = previousStop | |||
| 214 | + | }) | |||
| 209 | } | 215 | } | ||
| 210 | 216 | ||||
| 211 | func seamKillResult(requestID uint64, outcome protocol.KillOutcome, code uint16, text string) protocol.KillResult { | 217 | func seamKillResult(requestID uint64, outcome protocol.KillOutcome, code uint16, text string) protocol.KillResult { | ||
| @@ -215,6 +221,43 @@ func seamKillResult(requestID uint64, outcome protocol.KillOutcome, code uint16, | |||||
| 215 | // TestRunListRendersBrokerSessionsWithoutADaemon pins that runList reads the | 221 | // TestRunListRendersBrokerSessionsWithoutADaemon pins that runList reads the | ||
| 216 | // live session list through BrokerOperations over the per-user broker and | 222 | // live session list through BrokerOperations over the per-user broker and | ||
| 217 | // renders it, so a local list never reads durable state or dials a daemon. | 223 | // renders it, so a local list never reads durable state or dials a daemon. | ||
| 224 | + | func TestRunBrokerSnapshotListReportsNoDaemon(t *testing.T) { | |||
| 225 | + | remote := ports.BrokerDaemonObservation{ | |||
| 226 | + | Endpoint: "demo@host.test", DisplayOrigin: "demo@host.test", | |||
| 227 | + | Registration: domain.RemoteRegistration{Endpoint: "demo@host.test", Incarnation: [16]byte{1}, Generation: 1}, | |||
| 228 | + | Policy: seamBrokerTestPolicy(), Availability: domain.RemoteAvailabilityNoDaemon, | |||
| 229 | + | Sessions: []catalogue.RemoteCatalogSession{}, | |||
| 230 | + | } | |||
| 231 | + | snapshot := ports.BrokerSnapshot{Epoch: seamBrokerTestEpoch, Revision: 1, Daemons: []ports.BrokerDaemonObservation{remote}} | |||
| 232 | + | ||||
| 233 | + | var all bytes.Buffer | |||
| 234 | + | require.NoError(t, runBrokerSnapshotList(command{listAll: true}, snapshot, &all)) | |||
| 235 | + | require.Contains(t, all.String(), "demo@host.test: no vev daemon") | |||
| 236 | + | require.Contains(t, all.String(), "Enter in the picker to create a session") | |||
| 237 | + | require.NotContains(t, all.String(), "no sessions") | |||
| 238 | + | ||||
| 239 | + | var host bytes.Buffer | |||
| 240 | + | require.NoError(t, runBrokerSnapshotList(command{listHost: "demo@host.test"}, snapshot, &host)) | |||
| 241 | + | require.Contains(t, host.String(), "no vev daemon") | |||
| 242 | + | } | |||
| 243 | + | ||||
| 244 | + | func TestRunBrokerSnapshotListAllShowsNoDaemonAlongsideSessions(t *testing.T) { | |||
| 245 | + | remote := ports.BrokerDaemonObservation{ | |||
| 246 | + | Endpoint: "demo@host.test", DisplayOrigin: "demo@host.test", | |||
| 247 | + | Registration: domain.RemoteRegistration{Endpoint: "demo@host.test", Incarnation: [16]byte{1}, Generation: 1}, | |||
| 248 | + | Policy: seamBrokerTestPolicy(), Availability: domain.RemoteAvailabilityNoDaemon, | |||
| 249 | + | Sessions: []catalogue.RemoteCatalogSession{}, | |||
| 250 | + | } | |||
| 251 | + | local := seamBrokerTestSnapshot().Daemons[0] | |||
| 252 | + | local.InventoryKnown = true | |||
| 253 | + | local.Sessions = []catalogue.RemoteCatalogSession{{LifecycleID: domain.SessionLifecycleID{1}, Name: "work", State: catalogue.RemoteCatalogSessionUp, Tabs: []catalogue.RemoteCatalogTab{}}} | |||
| 254 | + | snapshot := ports.BrokerSnapshot{Epoch: seamBrokerTestEpoch, Revision: 1, Daemons: []ports.BrokerDaemonObservation{local, remote}} | |||
| 255 | + | var output bytes.Buffer | |||
| 256 | + | require.NoError(t, runBrokerSnapshotList(command{listAll: true}, snapshot, &output)) | |||
| 257 | + | require.Contains(t, output.String(), "demo@host.test: no vev daemon") | |||
| 258 | + | require.Contains(t, output.String(), "work") | |||
| 259 | + | } | |||
| 260 | + | ||||
| 218 | func TestRunListRendersBrokerSessionsWithoutADaemon(t *testing.T) { | 261 | func TestRunListRendersBrokerSessionsWithoutADaemon(t *testing.T) { | ||
| 219 | stream := newSeamBrokerStream(protocol.Sessions{Sessions: []protocol.SessionInfo{ | 262 | stream := newSeamBrokerStream(protocol.Sessions{Sessions: []protocol.SessionInfo{ | ||
| 220 | {Name: "work", State: protocol.SessionUp, Tabs: 2, Attached: true}, | 263 | {Name: "work", State: protocol.SessionUp, Tabs: 2, Attached: true}, | ||
M internal/app/run_test.go +6 -0
| @@ -534,6 +534,7 @@ func TestParseArgs(t *testing.T) { | |||
| 534 | 534 | wantHostTgt string | |
| 535 | 535 | wantAll bool | |
| 536 | 536 | wantDaemon bool | |
| 537 | + | wantBroker bool | |
| 537 | 538 | wantErr bool | |
| 538 | 539 | nonUsageErr bool | |
| 539 | 540 | }{ | |
| @@ -578,6 +579,8 @@ func TestParseArgs(t *testing.T) { | |||
| 578 | 579 | {name: "kill dashed name via terminator", args: []string{"kill", "--", "--all"}, wantKind: kindKill, wantName: "--all"}, | |
| 579 | 580 | {name: "kill all", args: []string{"kill", "--all"}, wantKind: kindKill, wantAll: true}, | |
| 580 | 581 | {name: "kill daemon", args: []string{"kill", "--daemon"}, wantKind: kindKill, wantDaemon: true}, | |
| 582 | + | {name: "kill broker", args: []string{"kill", "--broker"}, wantKind: kindKill, wantBroker: true}, | |
| 583 | + | {name: "kill broker rejects extra arg", args: []string{"kill", "--broker", "extra"}, wantErr: true}, | |
| 581 | 584 | {name: "kill without name", args: []string{"kill"}, wantErr: true}, | |
| 582 | 585 | {name: "kill terminator without name", args: []string{"kill", "--"}, wantErr: true}, | |
| 583 | 586 | {name: "kill all rejects extra arg", args: []string{"kill", "--all", "extra"}, wantErr: true}, | |
| @@ -654,6 +657,9 @@ func TestParseArgs(t *testing.T) { | |||
| 654 | 657 | if got.killDaemon != tt.wantDaemon { | |
| 655 | 658 | t.Errorf("killDaemon = %v, want %v", got.killDaemon, tt.wantDaemon) | |
| 656 | 659 | } | |
| 660 | + | if got.killBroker != tt.wantBroker { | |
| 661 | + | t.Errorf("killBroker = %v, want %v", got.killBroker, tt.wantBroker) | |
| 662 | + | } | |
| 657 | 663 | }) | |
| 658 | 664 | } | |
| 659 | 665 | } | |
| @@ -534,6 +534,7 @@ func TestParseArgs(t *testing.T) { | |||||
| 534 | wantHostTgt string | 534 | wantHostTgt string | ||
| 535 | wantAll bool | 535 | wantAll bool | ||
| 536 | wantDaemon bool | 536 | wantDaemon bool | ||
| 537 | + | wantBroker bool | |||
| 537 | wantErr bool | 538 | wantErr bool | ||
| 538 | nonUsageErr bool | 539 | nonUsageErr bool | ||
| 539 | }{ | 540 | }{ | ||
| @@ -578,6 +579,8 @@ func TestParseArgs(t *testing.T) { | |||||
| 578 | {name: "kill dashed name via terminator", args: []string{"kill", "--", "--all"}, wantKind: kindKill, wantName: "--all"}, | 579 | {name: "kill dashed name via terminator", args: []string{"kill", "--", "--all"}, wantKind: kindKill, wantName: "--all"}, | ||
| 579 | {name: "kill all", args: []string{"kill", "--all"}, wantKind: kindKill, wantAll: true}, | 580 | {name: "kill all", args: []string{"kill", "--all"}, wantKind: kindKill, wantAll: true}, | ||
| 580 | {name: "kill daemon", args: []string{"kill", "--daemon"}, wantKind: kindKill, wantDaemon: true}, | 581 | {name: "kill daemon", args: []string{"kill", "--daemon"}, wantKind: kindKill, wantDaemon: true}, | ||
| 582 | + | {name: "kill broker", args: []string{"kill", "--broker"}, wantKind: kindKill, wantBroker: true}, | |||
| 583 | + | {name: "kill broker rejects extra arg", args: []string{"kill", "--broker", "extra"}, wantErr: true}, | |||
| 581 | {name: "kill without name", args: []string{"kill"}, wantErr: true}, | 584 | {name: "kill without name", args: []string{"kill"}, wantErr: true}, | ||
| 582 | {name: "kill terminator without name", args: []string{"kill", "--"}, wantErr: true}, | 585 | {name: "kill terminator without name", args: []string{"kill", "--"}, wantErr: true}, | ||
| 583 | {name: "kill all rejects extra arg", args: []string{"kill", "--all", "extra"}, wantErr: true}, | 586 | {name: "kill all rejects extra arg", args: []string{"kill", "--all", "extra"}, wantErr: true}, | ||
| @@ -654,6 +657,9 @@ func TestParseArgs(t *testing.T) { | |||||
| 654 | if got.killDaemon != tt.wantDaemon { | 657 | if got.killDaemon != tt.wantDaemon { | ||
| 655 | t.Errorf("killDaemon = %v, want %v", got.killDaemon, tt.wantDaemon) | 658 | t.Errorf("killDaemon = %v, want %v", got.killDaemon, tt.wantDaemon) | ||
| 656 | } | 659 | } | ||
| 660 | + | if got.killBroker != tt.wantBroker { | |||
| 661 | + | t.Errorf("killBroker = %v, want %v", got.killBroker, tt.wantBroker) | |||
| 662 | + | } | |||
| 657 | }) | 663 | }) | ||
| 658 | } | 664 | } | ||
| 659 | } | 665 | } | ||