fix(app): stop an existing daemon directly when no broker runs
- commit
- f2870bb23f61ff96f64dc26b57d3115f6a0238b7
- author
- brice · · 2026-09-23 11:44 UTC
- parent
- 31711a6
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +109 -19
M internal/app/list_readonly_test.go +59 -0
| @@ -4,16 +4,22 @@ import ( | |||
| 4 | 4 | "bytes" | |
| 5 | 5 | "context" | |
| 6 | 6 | "errors" | |
| 7 | + | "net" | |
| 7 | 8 | "os" | |
| 8 | 9 | "path/filepath" | |
| 10 | + | "syscall" | |
| 9 | 11 | "testing" | |
| 10 | 12 | "time" | |
| 11 | 13 | ||
| 12 | 14 | "github.com/stretchr/testify/require" | |
| 13 | 15 | ||
| 16 | + | "github.com/bnema/vev/internal/adapters/ipc" | |
| 17 | + | "github.com/bnema/vev/internal/adapters/sessionwire" | |
| 14 | 18 | "github.com/bnema/vev/internal/domain" | |
| 15 | 19 | "github.com/bnema/vev/internal/ports" | |
| 20 | + | "github.com/bnema/vev/internal/protocol" | |
| 16 | 21 | "github.com/bnema/vev/internal/protocol/catalogue" | |
| 22 | + | "github.com/bnema/vev/internal/protocol/wire" | |
| 17 | 23 | ) | |
| 18 | 24 | ||
| 19 | 25 | func TestObservedHostStatus(t *testing.T) { | |
| @@ -107,3 +113,56 @@ func TestHostListUsesExistingConnector(t *testing.T) { | |||
| 107 | 113 | require.True(t, called) | |
| 108 | 114 | require.Equal(t, "no hosts\n", output.String()) | |
| 109 | 115 | } | |
| 116 | + | ||
| 117 | + | // TestStopDaemonWithoutBrokerReachesExistingDaemon pins that a stop request | |
| 118 | + | // with no broker running goes straight to an existing local daemon: the idle | |
| 119 | + | // broker exits long before the daemon, so its absence never implies that no | |
| 120 | + | // daemon runs, and nothing is started to find out. | |
| 121 | + | func TestStopDaemonWithoutBrokerReachesExistingDaemon(t *testing.T) { | |
| 122 | + | tests := []struct { | |
| 123 | + | name string | |
| 124 | + | daemon bool | |
| 125 | + | outcome protocol.KillOutcome | |
| 126 | + | wantErr error | |
| 127 | + | }{ | |
| 128 | + | {name: "running daemon stops", daemon: true, outcome: protocol.KillSucceeded}, | |
| 129 | + | {name: "absent daemon is reported, not started", daemon: false, wantErr: errDaemonNotRunning}, | |
| 130 | + | } | |
| 131 | + | for _, tt := range tests { | |
| 132 | + | t.Run(tt.name, func(t *testing.T) { | |
| 133 | + | received := make(chan protocol.ClientMessage, 1) | |
| 134 | + | previous := dialListDaemon | |
| 135 | + | t.Cleanup(func() { dialListDaemon = previous }) | |
| 136 | + | dialListDaemon = func(context.Context, string, ...ipc.Option) (wire.Transport, error) { | |
| 137 | + | if !tt.daemon { | |
| 138 | + | return nil, &net.OpError{Op: "dial", Net: "unix", Err: syscall.ENOENT} | |
| 139 | + | } | |
| 140 | + | clientSide, serverSide := net.Pipe() | |
| 141 | + | server := sessionwire.NewServerConnection(ipc.NewTransport(serverSide)) | |
| 142 | + | go func() { | |
| 143 | + | defer func() { _ = server.Close() }() | |
| 144 | + | message, err := server.ReceiveClient() | |
| 145 | + | if err != nil { | |
| 146 | + | return | |
| 147 | + | } | |
| 148 | + | received <- message | |
| 149 | + | _ = server.SendServer(protocol.KillResult{RequestID: 1, Outcome: tt.outcome}) | |
| 150 | + | }() | |
| 151 | + | return ipc.NewTransport(clientSide), nil | |
| 152 | + | } | |
| 153 | + | ||
| 154 | + | err := stopDaemonWithoutBroker(context.Background()) | |
| 155 | + | if tt.wantErr != nil { | |
| 156 | + | require.ErrorIs(t, err, tt.wantErr) | |
| 157 | + | return | |
| 158 | + | } | |
| 159 | + | require.NoError(t, err) | |
| 160 | + | select { | |
| 161 | + | case message := <-received: | |
| 162 | + | require.Equal(t, protocol.Kill{RequestID: 1, Scope: protocol.KillDaemon}, message) | |
| 163 | + | case <-time.After(5 * time.Second): | |
| 164 | + | t.Fatal("the daemon never received the stop request") | |
| 165 | + | } | |
| 166 | + | }) | |
| 167 | + | } | |
| 168 | + | } | |
| @@ -4,16 +4,22 @@ import ( | |||||
| 4 | "bytes" | 4 | "bytes" | ||
| 5 | "context" | 5 | "context" | ||
| 6 | "errors" | 6 | "errors" | ||
| 7 | + | "net" | |||
| 7 | "os" | 8 | "os" | ||
| 8 | "path/filepath" | 9 | "path/filepath" | ||
| 10 | + | "syscall" | |||
| 9 | "testing" | 11 | "testing" | ||
| 10 | "time" | 12 | "time" | ||
| 11 | 13 | ||||
| 12 | "github.com/stretchr/testify/require" | 14 | "github.com/stretchr/testify/require" | ||
| 13 | 15 | ||||
| 16 | + | "github.com/bnema/vev/internal/adapters/ipc" | |||
| 17 | + | "github.com/bnema/vev/internal/adapters/sessionwire" | |||
| 14 | "github.com/bnema/vev/internal/domain" | 18 | "github.com/bnema/vev/internal/domain" | ||
| 15 | "github.com/bnema/vev/internal/ports" | 19 | "github.com/bnema/vev/internal/ports" | ||
| 20 | + | "github.com/bnema/vev/internal/protocol" | |||
| 16 | "github.com/bnema/vev/internal/protocol/catalogue" | 21 | "github.com/bnema/vev/internal/protocol/catalogue" | ||
| 22 | + | "github.com/bnema/vev/internal/protocol/wire" | |||
| 17 | ) | 23 | ) | ||
| 18 | 24 | ||||
| 19 | func TestObservedHostStatus(t *testing.T) { | 25 | func TestObservedHostStatus(t *testing.T) { | ||
| @@ -107,3 +113,56 @@ func TestHostListUsesExistingConnector(t *testing.T) { | |||||
| 107 | require.True(t, called) | 113 | require.True(t, called) | ||
| 108 | require.Equal(t, "no hosts\n", output.String()) | 114 | require.Equal(t, "no hosts\n", output.String()) | ||
| 109 | } | 115 | } | ||
| 116 | + | ||||
| 117 | + | // TestStopDaemonWithoutBrokerReachesExistingDaemon pins that a stop request | |||
| 118 | + | // with no broker running goes straight to an existing local daemon: the idle | |||
| 119 | + | // broker exits long before the daemon, so its absence never implies that no | |||
| 120 | + | // daemon runs, and nothing is started to find out. | |||
| 121 | + | func TestStopDaemonWithoutBrokerReachesExistingDaemon(t *testing.T) { | |||
| 122 | + | tests := []struct { | |||
| 123 | + | name string | |||
| 124 | + | daemon bool | |||
| 125 | + | outcome protocol.KillOutcome | |||
| 126 | + | wantErr error | |||
| 127 | + | }{ | |||
| 128 | + | {name: "running daemon stops", daemon: true, outcome: protocol.KillSucceeded}, | |||
| 129 | + | {name: "absent daemon is reported, not started", daemon: false, wantErr: errDaemonNotRunning}, | |||
| 130 | + | } | |||
| 131 | + | for _, tt := range tests { | |||
| 132 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 133 | + | received := make(chan protocol.ClientMessage, 1) | |||
| 134 | + | previous := dialListDaemon | |||
| 135 | + | t.Cleanup(func() { dialListDaemon = previous }) | |||
| 136 | + | dialListDaemon = func(context.Context, string, ...ipc.Option) (wire.Transport, error) { | |||
| 137 | + | if !tt.daemon { | |||
| 138 | + | return nil, &net.OpError{Op: "dial", Net: "unix", Err: syscall.ENOENT} | |||
| 139 | + | } | |||
| 140 | + | clientSide, serverSide := net.Pipe() | |||
| 141 | + | server := sessionwire.NewServerConnection(ipc.NewTransport(serverSide)) | |||
| 142 | + | go func() { | |||
| 143 | + | defer func() { _ = server.Close() }() | |||
| 144 | + | message, err := server.ReceiveClient() | |||
| 145 | + | if err != nil { | |||
| 146 | + | return | |||
| 147 | + | } | |||
| 148 | + | received <- message | |||
| 149 | + | _ = server.SendServer(protocol.KillResult{RequestID: 1, Outcome: tt.outcome}) | |||
| 150 | + | }() | |||
| 151 | + | return ipc.NewTransport(clientSide), nil | |||
| 152 | + | } | |||
| 153 | + | ||||
| 154 | + | err := stopDaemonWithoutBroker(context.Background()) | |||
| 155 | + | if tt.wantErr != nil { | |||
| 156 | + | require.ErrorIs(t, err, tt.wantErr) | |||
| 157 | + | return | |||
| 158 | + | } | |||
| 159 | + | require.NoError(t, err) | |||
| 160 | + | select { | |||
| 161 | + | case message := <-received: | |||
| 162 | + | require.Equal(t, protocol.Kill{RequestID: 1, Scope: protocol.KillDaemon}, message) | |||
| 163 | + | case <-time.After(5 * time.Second): | |||
| 164 | + | t.Fatal("the daemon never received the stop request") | |||
| 165 | + | } | |||
| 166 | + | }) | |||
| 167 | + | } | |||
| 168 | + | } | |||
M internal/app/run.go +50 -19
| @@ -1032,7 +1032,7 @@ func requestDaemonStop(ctx context.Context) error { | |||
| 1032 | 1032 | service, err := connectDaemonStopBroker(ctx) | |
| 1033 | 1033 | if err != nil { | |
| 1034 | 1034 | if errors.Is(err, errBrokerAbsent) { | |
| 1035 | - | return errDaemonNotRunning | |
| 1035 | + | return stopDaemonWithoutBroker(ctx) | |
| 1036 | 1036 | } | |
| 1037 | 1037 | return unreachableBrokerError(err) | |
| 1038 | 1038 | } | |
| @@ -1336,41 +1336,72 @@ func runList(ctx context.Context, cmd command) error { | |||
| 1336 | 1336 | var connectListBroker = connectExistingBroker | |
| 1337 | 1337 | var dialListDaemon = ipc.DialContext | |
| 1338 | 1338 | ||
| 1339 | + | // errNoLocalDaemon reports that no local daemon socket answers; nothing was | |
| 1340 | + | // started to find out. | |
| 1341 | + | var errNoLocalDaemon = errors.New("vev: no local daemon") | |
| 1342 | + | ||
| 1343 | + | // exchangeLocalDaemon sends one request directly to an existing local daemon | |
| 1344 | + | // and returns its reply. It is the broker-less path of read and stop | |
| 1345 | + | // commands: the broker being absent is the normal idle state, and these | |
| 1346 | + | // commands must stay true without starting it. It never spawns a daemon. | |
| 1347 | + | func exchangeLocalDaemon(ctx context.Context, request protocol.ClientMessage) (protocol.ServerMessage, error) { | |
| 1348 | + | transport, err := dialListDaemon(ctx, ipc.SocketDir()) | |
| 1349 | + | if err != nil { | |
| 1350 | + | if backendAbsent(err) { | |
| 1351 | + | return nil, errNoLocalDaemon | |
| 1352 | + | } | |
| 1353 | + | return nil, err | |
| 1354 | + | } | |
| 1355 | + | connection := sessionwire.NewClientConnection(transport) | |
| 1356 | + | defer func() { _ = connection.Close() }() | |
| 1357 | + | if err := connection.SendClient(request); err != nil { | |
| 1358 | + | return nil, err | |
| 1359 | + | } | |
| 1360 | + | return connection.ReceiveServer() | |
| 1361 | + | } | |
| 1362 | + | ||
| 1339 | 1363 | func runListWithoutBroker(ctx context.Context, cmd command, out io.Writer) error { | |
| 1340 | 1364 | if cmd.listHost != "" { | |
| 1341 | 1365 | _, err := fmt.Fprintf(out, "%s: not connected (no broker running)\n", cmd.listHost) | |
| 1342 | 1366 | return err | |
| 1343 | 1367 | } | |
| 1344 | - | transport, err := dialListDaemon(ctx, ipc.SocketDir()) | |
| 1345 | - | if err != nil { | |
| 1346 | - | if !backendAbsent(err) { | |
| 1347 | - | return fmt.Errorf("vev: list local sessions: %w", err) | |
| 1348 | - | } | |
| 1349 | - | } else { | |
| 1350 | - | connection := sessionwire.NewClientConnection(transport) | |
| 1351 | - | defer func() { _ = connection.Close() }() | |
| 1352 | - | if err := connection.SendClient(protocol.List{}); err != nil { | |
| 1353 | - | return fmt.Errorf("vev: request local sessions: %w", err) | |
| 1354 | - | } | |
| 1355 | - | reply, err := connection.ReceiveServer() | |
| 1356 | - | if err != nil { | |
| 1357 | - | return fmt.Errorf("vev: read local sessions: %w", err) | |
| 1358 | - | } | |
| 1368 | + | reply, err := exchangeLocalDaemon(ctx, protocol.List{}) | |
| 1369 | + | switch { | |
| 1370 | + | case errors.Is(err, errNoLocalDaemon): | |
| 1371 | + | printSessions(out, nil) | |
| 1372 | + | case err != nil: | |
| 1373 | + | return fmt.Errorf("vev: list local sessions: %w", err) | |
| 1374 | + | default: | |
| 1359 | 1375 | sessions, ok := reply.(protocol.Sessions) | |
| 1360 | 1376 | if !ok { | |
| 1361 | 1377 | return fmt.Errorf("vev: unexpected local list reply %T", reply) | |
| 1362 | 1378 | } | |
| 1363 | 1379 | printSessions(out, sessions.Sessions) | |
| 1364 | 1380 | } | |
| 1365 | - | if err != nil { | |
| 1366 | - | printSessions(out, nil) | |
| 1367 | - | } | |
| 1368 | 1381 | if cmd.listAll { | |
| 1369 | 1382 | return listDisconnectedHosts(out) | |
| 1370 | 1383 | } | |
| 1371 | 1384 | return nil | |
| 1372 | 1385 | } | |
| 1373 | 1386 | ||
| 1387 | + | // stopDaemonWithoutBroker asks an existing local daemon to stop directly. A | |
| 1388 | + | // daemon routinely outlives the idle broker, so "no broker" never means "no | |
| 1389 | + | // daemon". | |
| 1390 | + | func stopDaemonWithoutBroker(ctx context.Context) error { | |
| 1391 | + | reply, err := exchangeLocalDaemon(ctx, protocol.Kill{RequestID: 1, Scope: protocol.KillDaemon}) | |
| 1392 | + | if errors.Is(err, errNoLocalDaemon) { | |
| 1393 | + | return errDaemonNotRunning | |
| 1394 | + | } | |
| 1395 | + | if err != nil { | |
| 1396 | + | return fmt.Errorf("vev: stop local daemon: %w", err) | |
| 1397 | + | } | |
| 1398 | + | result, ok := reply.(protocol.KillResult) | |
| 1399 | + | if !ok { | |
| 1400 | + | return fmt.Errorf("vev: unexpected daemon stop reply %T", reply) | |
| 1401 | + | } | |
| 1402 | + | return brokerKillResultError(result) | |
| 1403 | + | } | |
| 1404 | + | ||
| 1374 | 1405 | // printSessions renders a session table (or a friendly note when empty). | |
| 1375 | 1406 | func printSessions(w io.Writer, sessions []protocol.SessionInfo) { | |
| 1376 | 1407 | if len(sessions) == 0 { | |
| @@ -1032,7 +1032,7 @@ func requestDaemonStop(ctx context.Context) error { | |||||
| 1032 | service, err := connectDaemonStopBroker(ctx) | 1032 | service, err := connectDaemonStopBroker(ctx) | ||
| 1033 | if err != nil { | 1033 | if err != nil { | ||
| 1034 | if errors.Is(err, errBrokerAbsent) { | 1034 | if errors.Is(err, errBrokerAbsent) { | ||
| 1035 | - | return errDaemonNotRunning | 1035 | + | return stopDaemonWithoutBroker(ctx) |
| 1036 | } | 1036 | } | ||
| 1037 | return unreachableBrokerError(err) | 1037 | return unreachableBrokerError(err) | ||
| 1038 | } | 1038 | } | ||
| @@ -1336,41 +1336,72 @@ func runList(ctx context.Context, cmd command) error { | |||||
| 1336 | var connectListBroker = connectExistingBroker | 1336 | var connectListBroker = connectExistingBroker | ||
| 1337 | var dialListDaemon = ipc.DialContext | 1337 | var dialListDaemon = ipc.DialContext | ||
| 1338 | 1338 | ||||
| 1339 | + | // errNoLocalDaemon reports that no local daemon socket answers; nothing was | |||
| 1340 | + | // started to find out. | |||
| 1341 | + | var errNoLocalDaemon = errors.New("vev: no local daemon") | |||
| 1342 | + | ||||
| 1343 | + | // exchangeLocalDaemon sends one request directly to an existing local daemon | |||
| 1344 | + | // and returns its reply. It is the broker-less path of read and stop | |||
| 1345 | + | // commands: the broker being absent is the normal idle state, and these | |||
| 1346 | + | // commands must stay true without starting it. It never spawns a daemon. | |||
| 1347 | + | func exchangeLocalDaemon(ctx context.Context, request protocol.ClientMessage) (protocol.ServerMessage, error) { | |||
| 1348 | + | transport, err := dialListDaemon(ctx, ipc.SocketDir()) | |||
| 1349 | + | if err != nil { | |||
| 1350 | + | if backendAbsent(err) { | |||
| 1351 | + | return nil, errNoLocalDaemon | |||
| 1352 | + | } | |||
| 1353 | + | return nil, err | |||
| 1354 | + | } | |||
| 1355 | + | connection := sessionwire.NewClientConnection(transport) | |||
| 1356 | + | defer func() { _ = connection.Close() }() | |||
| 1357 | + | if err := connection.SendClient(request); err != nil { | |||
| 1358 | + | return nil, err | |||
| 1359 | + | } | |||
| 1360 | + | return connection.ReceiveServer() | |||
| 1361 | + | } | |||
| 1362 | + | ||||
| 1339 | func runListWithoutBroker(ctx context.Context, cmd command, out io.Writer) error { | 1363 | func runListWithoutBroker(ctx context.Context, cmd command, out io.Writer) error { | ||
| 1340 | if cmd.listHost != "" { | 1364 | if cmd.listHost != "" { | ||
| 1341 | _, err := fmt.Fprintf(out, "%s: not connected (no broker running)\n", cmd.listHost) | 1365 | _, err := fmt.Fprintf(out, "%s: not connected (no broker running)\n", cmd.listHost) | ||
| 1342 | return err | 1366 | return err | ||
| 1343 | } | 1367 | } | ||
| 1344 | - | transport, err := dialListDaemon(ctx, ipc.SocketDir()) | 1368 | + | reply, err := exchangeLocalDaemon(ctx, protocol.List{}) |
| 1345 | - | if err != nil { | 1369 | + | switch { |
| 1346 | - | if !backendAbsent(err) { | 1370 | + | case errors.Is(err, errNoLocalDaemon): |
| 1347 | - | return fmt.Errorf("vev: list local sessions: %w", err) | 1371 | + | printSessions(out, nil) |
| 1348 | - | } | 1372 | + | case err != nil: |
| 1349 | - | } else { | 1373 | + | return fmt.Errorf("vev: list local sessions: %w", err) |
| 1350 | - | connection := sessionwire.NewClientConnection(transport) | 1374 | + | default: |
| 1351 | - | defer func() { _ = connection.Close() }() | |||
| 1352 | - | if err := connection.SendClient(protocol.List{}); err != nil { | |||
| 1353 | - | return fmt.Errorf("vev: request local sessions: %w", err) | |||
| 1354 | - | } | |||
| 1355 | - | reply, err := connection.ReceiveServer() | |||
| 1356 | - | if err != nil { | |||
| 1357 | - | return fmt.Errorf("vev: read local sessions: %w", err) | |||
| 1358 | - | } | |||
| 1359 | sessions, ok := reply.(protocol.Sessions) | 1375 | sessions, ok := reply.(protocol.Sessions) | ||
| 1360 | if !ok { | 1376 | if !ok { | ||
| 1361 | return fmt.Errorf("vev: unexpected local list reply %T", reply) | 1377 | return fmt.Errorf("vev: unexpected local list reply %T", reply) | ||
| 1362 | } | 1378 | } | ||
| 1363 | printSessions(out, sessions.Sessions) | 1379 | printSessions(out, sessions.Sessions) | ||
| 1364 | } | 1380 | } | ||
| 1365 | - | if err != nil { | |||
| 1366 | - | printSessions(out, nil) | |||
| 1367 | - | } | |||
| 1368 | if cmd.listAll { | 1381 | if cmd.listAll { | ||
| 1369 | return listDisconnectedHosts(out) | 1382 | return listDisconnectedHosts(out) | ||
| 1370 | } | 1383 | } | ||
| 1371 | return nil | 1384 | return nil | ||
| 1372 | } | 1385 | } | ||
| 1373 | 1386 | ||||
| 1387 | + | // stopDaemonWithoutBroker asks an existing local daemon to stop directly. A | |||
| 1388 | + | // daemon routinely outlives the idle broker, so "no broker" never means "no | |||
| 1389 | + | // daemon". | |||
| 1390 | + | func stopDaemonWithoutBroker(ctx context.Context) error { | |||
| 1391 | + | reply, err := exchangeLocalDaemon(ctx, protocol.Kill{RequestID: 1, Scope: protocol.KillDaemon}) | |||
| 1392 | + | if errors.Is(err, errNoLocalDaemon) { | |||
| 1393 | + | return errDaemonNotRunning | |||
| 1394 | + | } | |||
| 1395 | + | if err != nil { | |||
| 1396 | + | return fmt.Errorf("vev: stop local daemon: %w", err) | |||
| 1397 | + | } | |||
| 1398 | + | result, ok := reply.(protocol.KillResult) | |||
| 1399 | + | if !ok { | |||
| 1400 | + | return fmt.Errorf("vev: unexpected daemon stop reply %T", reply) | |||
| 1401 | + | } | |||
| 1402 | + | return brokerKillResultError(result) | |||
| 1403 | + | } | |||
| 1404 | + | ||||
| 1374 | // printSessions renders a session table (or a friendly note when empty). | 1405 | // printSessions renders a session table (or a friendly note when empty). | ||
| 1375 | func printSessions(w io.Writer, sessions []protocol.SessionInfo) { | 1406 | func printSessions(w io.Writer, sessions []protocol.SessionInfo) { | ||
| 1376 | if len(sessions) == 0 { | 1407 | if len(sessions) == 0 { | ||