1//go:build linux
2
3package app
4
5import (
6 "bytes"
7 "context"
8 "errors"
9 "fmt"
10 "io"
11 "net"
12 "os"
13 "path/filepath"
14 "strconv"
15 "strings"
16 "syscall"
17 "time"
18
19 "github.com/bnema/vev/internal/adapters/brokeripc"
20 "github.com/bnema/vev/internal/adapters/ipc"
21 "github.com/bnema/vev/internal/adapters/lifecycle"
22 "github.com/bnema/vev/internal/platform"
23)
24
25const (
26 // killAllStopTimeout bounds how long each wave waits after SIGTERM
27 // before it escalates to SIGKILL.
28 killAllStopTimeout = 5 * time.Second
29 // killAllRounds caps how many scan-and-stop rounds run, so a process that
30 // a stopping one respawned is caught by the next scan.
31 killAllRounds = 3
32)
33
34// vevProcess is one running process of the installed vev binary.
35type vevProcess struct {
36 pid int
37 role string // "client", "broker", "daemon", or "helper"
38}
39
40// killAllScope selects which processes `kill --all` may stop: this user's
41// processes of the installed binary that use the same vev socket directory,
42// so VEV_ENV and XDG-sandboxed runs never touch each other.
43type killAllScope struct {
44 procRoot string
45 exe string
46 uid int
47 self int
48 socketDir string // ipc.SocketDir of the caller
49}
50
51// killAllReport is what one `kill --all` did.
52type killAllReport struct {
53 stopped []vevProcess
54 forced []vevProcess // needed SIGKILL
55 removed []string
56 errs []error
57}
58
59// runKillAll stops every vev process of this environment, including builds an
60// upgrade replaced on disk, then removes the sockets they left behind. It needs
61// no broker, so it also works when the running broker is from another build.
62// The daemon checkpoints named sessions on SIGTERM; numbered sessions close.
63func runKillAll(ctx context.Context, out io.Writer) error {
64 if os.Getenv("VEV") != "" {
65 return errors.New("`kill --all` stops the vev you are running in; run it from a terminal outside vev")
66 }
67 exe, err := os.Executable()
68 if err != nil {
69 return fmt.Errorf("resolve executable: %w", err)
70 }
71 scope := killAllScope{
72 procRoot: "/proc",
73 exe: installedPath(exe),
74 uid: os.Getuid(),
75 self: os.Getpid(),
76 socketDir: ipc.SocketDir(),
77 }
78 report := stopEverything(ctx, scope, killAllStopTimeout)
79 printKillAllReport(out, report)
80 return errors.Join(report.errs...)
81}
82
83// stopEverything runs scan-and-stop rounds until a scan finds nothing, then
84// removes dead sockets. A process that already failed to die is not retried.
85func stopEverything(ctx context.Context, scope killAllScope, timeout time.Duration) killAllReport {
86 var report killAllReport
87 failed := map[int]bool{}
88 scan := func() ([]vevProcess, error) {
89 procs, err := findVevProcesses(scope)
90 live := procs[:0]
91 for _, p := range procs {
92 if !failed[p.pid] {
93 live = append(live, p)
94 }
95 }
96 return live, err
97 }
98 for round := 0; ; round++ {
99 procs, err := scan()
100 if err != nil {
101 report.errs = append(report.errs, err)
102 return report
103 }
104 if len(procs) == 0 {
105 if len(failed) == 0 {
106 report.removed = removeDeadSockets(scope.socketDir)
107 }
108 return report
109 }
110 if round == killAllRounds {
111 report.errs = append(report.errs, fmt.Errorf("%d vev processes kept starting again; run `vev kill --all` once more", len(procs)))
112 return report
113 }
114 // Clients first, so none of them can start a broker or daemon from
115 // its old binary while the rest stop.
116 var clients, services []vevProcess
117 for _, p := range procs {
118 if p.role == "client" {
119 clients = append(clients, p)
120 } else {
121 services = append(services, p)
122 }
123 }
124 for _, wave := range [][]vevProcess{clients, services} {
125 for _, p := range stopWave(ctx, wave, timeout, &report) {
126 failed[p.pid] = true
127 }
128 }
129 if ctx.Err() != nil {
130 report.errs = append(report.errs, ctx.Err())
131 return report
132 }
133 }
134}
135
136// installedPath is the on-disk path an executable was started from, without
137// the " (deleted)" suffix the kernel adds once an upgrade replaced it.
138func installedPath(exe string) string {
139 return strings.TrimSuffix(exe, " (deleted)")
140}
141
142// findVevProcesses lists the processes in scope. The caller never matches
143// itself.
144func findVevProcesses(scope killAllScope) ([]vevProcess, error) {
145 entries, err := os.ReadDir(scope.procRoot)
146 if err != nil {
147 return nil, fmt.Errorf("list processes: %w", err)
148 }
149 var procs []vevProcess
150 for _, entry := range entries {
151 pid, err := strconv.Atoi(entry.Name())
152 if err != nil || pid == scope.self {
153 continue
154 }
155 dir := filepath.Join(scope.procRoot, entry.Name())
156 info, err := os.Stat(dir)
157 if err != nil {
158 continue // the process exited meanwhile
159 }
160 if stat, ok := info.Sys().(*syscall.Stat_t); !ok || int(stat.Uid) != scope.uid {
161 continue
162 }
163 target, err := os.Readlink(filepath.Join(dir, "exe"))
164 if err != nil || installedPath(target) != scope.exe {
165 continue
166 }
167 if processSocketDir(dir, scope.uid) != scope.socketDir {
168 continue // another VEV_ENV or sandbox
169 }
170 cmdline, err := os.ReadFile(filepath.Join(dir, "cmdline"))
171 if err != nil {
172 continue
173 }
174 procs = append(procs, vevProcess{pid: pid, role: processRole(cmdline)})
175 }
176 return procs, nil
177}
178
179// processSocketDir resolves the vev socket directory of the process in
180// procDir from its initial environment and working directory, the same way
181// vev itself does after VEV_ENV activation. It returns "" when unreadable.
182func processSocketDir(procDir string, uid int) string {
183 environ, err := os.ReadFile(filepath.Join(procDir, "environ"))
184 if err != nil {
185 return ""
186 }
187 env := map[string]string{}
188 for _, entry := range strings.Split(string(environ), "\x00") {
189 if key, value, ok := strings.Cut(entry, "="); ok {
190 env[key] = value
191 }
192 }
193 cwd, _ := os.Readlink(filepath.Join(procDir, "cwd"))
194 // Like os.Getwd, prefer $PWD when it names the same directory, so a repo
195 // opened through a symlink resolves to the path vev itself used.
196 if pwd := env["PWD"]; filepath.IsAbs(pwd) {
197 pwdInfo, pwdErr := os.Stat(pwd)
198 cwdInfo, cwdErr := os.Stat(filepath.Join(procDir, "cwd"))
199 if pwdErr == nil && cwdErr == nil && os.SameFile(pwdInfo, cwdInfo) {
200 cwd = pwd
201 }
202 }
203 runtime := platform.RuntimeDirFor(func(key string) string { return env[key] }, cwd)
204 return ipc.SocketDirFor(runtime, uid)
205}
206
207// processRole names a vev process from its NUL-separated command line.
208func processRole(cmdline []byte) string {
209 args := strings.Split(strings.TrimRight(string(cmdline), "\x00"), "\x00")
210 if len(args) < 2 {
211 return "client"
212 }
213 switch args[1] {
214 case "--daemon":
215 return "daemon"
216 case productionBrokerServeCommand:
217 return "broker"
218 case "--daemon-launcher", "--web-daemon", "--web-serve", "--ui-driver":
219 return "helper"
220 }
221 if strings.HasPrefix(args[1], "_") {
222 return "helper"
223 }
224 return "client"
225}
226
227// stopWave sends SIGTERM to every process, waits up to timeout for all of them
228// to exit, then sends SIGKILL to the ones still running. One stuck process
229// never stops the rest of the wave; the ones that survive are returned.
230func stopWave(ctx context.Context, procs []vevProcess, timeout time.Duration, report *killAllReport) (survivors []vevProcess) {
231 for _, p := range procs {
232 _ = syscall.Kill(p.pid, syscall.SIGTERM)
233 }
234 deadline := time.Now().Add(timeout)
235 for _, p := range procs {
236 if waitProcessExit(ctx, p.pid, max(time.Until(deadline), 0)) == nil {
237 report.stopped = append(report.stopped, p)
238 continue
239 }
240 _ = syscall.Kill(p.pid, syscall.SIGKILL)
241 if err := waitProcessExit(context.WithoutCancel(ctx), p.pid, time.Second); err != nil {
242 report.errs = append(report.errs, fmt.Errorf("%s (pid %d) did not stop: %w", p.role, p.pid, err))
243 survivors = append(survivors, p)
244 continue
245 }
246 report.stopped = append(report.stopped, p)
247 report.forced = append(report.forced, p)
248 }
249 return survivors
250}
251
252// removeDeadSockets deletes vev sockets nobody listens on any more, so the
253// next start never trips over a leftover endpoint. It holds each lifecycle
254// lock while it checks, so a daemon or broker that is starting keeps its
255// socket, and it never removes anything that is not a socket.
256func removeDeadSockets(dir string) []string {
257 var removed []string
258 for _, runtime := range []struct{ lockDir, socket string }{
259 {dir, filepath.Join(dir, "daemon.sock")},
260 {dir, filepath.Join(dir, "daemonmux.sock")},
261 {filepath.Join(dir, "broker"), brokeripc.SocketPath(filepath.Join(dir, "broker"))},
262 } {
263 info, err := os.Lstat(runtime.socket)
264 if err != nil || info.Mode().Type() != os.ModeSocket {
265 continue
266 }
267 owner, err := lifecycle.TryAcquire(runtime.lockDir)
268 if err != nil {
269 continue // busy: a process is starting or still owns it
270 }
271 conn, err := net.DialTimeout("unix", runtime.socket, 200*time.Millisecond)
272 if err == nil {
273 _ = conn.Close()
274 } else if errors.Is(err, syscall.ECONNREFUSED) && os.Remove(runtime.socket) == nil {
275 removed = append(removed, runtime.socket)
276 }
277 _ = owner.Release()
278 }
279 return removed
280}
281
282func printKillAllReport(out io.Writer, report killAllReport) {
283 if len(report.stopped) == 0 && len(report.removed) == 0 && len(report.errs) == 0 {
284 _, _ = fmt.Fprintln(out, "nothing to stop: vev was not running")
285 return
286 }
287 for _, p := range report.stopped {
288 _, _ = fmt.Fprintf(out, "stopped %s (pid %d)\n", p.role, p.pid)
289 }
290 for _, path := range report.removed {
291 _, _ = fmt.Fprintf(out, "removed stale socket %s\n", path)
292 }
293 for _, p := range report.forced {
294 if p.role == "daemon" {
295 _, _ = fmt.Fprintf(out, "warning: the daemon (pid %d) had to be force-killed; recent output in named sessions may be lost\n", p.pid)
296 }
297 }
298 if len(report.errs) > 0 {
299 return
300 }
301 _, _ = fmt.Fprintln(out, "vev is fully stopped; named sessions come back on the next start, numbered sessions are closed")
302}
303
304// waitProcessExit polls until pid no longer exists or is a zombie: a dead
305// process whose parent (for example a shell) has not reaped it yet.
306func waitProcessExit(ctx context.Context, pid int, timeout time.Duration) error {
307 ctx, cancel := context.WithTimeout(ctx, timeout)
308 defer cancel()
309 ticker := time.NewTicker(20 * time.Millisecond)
310 defer ticker.Stop()
311 for {
312 if err := syscall.Kill(pid, 0); errors.Is(err, syscall.ESRCH) || isZombie(pid) {
313 return nil
314 }
315 select {
316 case <-ctx.Done():
317 return fmt.Errorf("still running after %s: %w", timeout, ctx.Err())
318 case <-ticker.C:
319 }
320 }
321}
322
323// isZombie reports whether /proc says pid has exited but is not reaped yet.
324func isZombie(pid int) bool {
325 stat, err := os.ReadFile(filepath.Join("/proc", strconv.Itoa(pid), "stat"))
326 if err != nil {
327 return false
328 }
329 // The state follows the parenthesised command name, which may hold spaces.
330 end := bytes.LastIndexByte(stat, ')')
331 return end >= 0 && end+2 < len(stat) && stat[end+2] == 'Z'
332}