diff --git a/.changes/windows-event-bus-lifecycle.md b/.changes/windows-event-bus-lifecycle.md new file mode 100644 index 00000000..f158d0ad --- /dev/null +++ b/.changes/windows-event-bus-lifecycle.md @@ -0,0 +1,5 @@ +--- +category: Fixed +--- + +- **Windows event bus lifecycle** — start event consumers without unsupported inherited file descriptors, stop buses through local IPC with a termination fallback, and preserve subscription cleanup when startup fails. diff --git a/internal/app/event_command.go b/internal/app/event_command.go index 533c95a3..f0cc1e24 100644 --- a/internal/app/event_command.go +++ b/internal/app/event_command.go @@ -1220,7 +1220,8 @@ func newEventStopCommandWithFlags(globalFlags ...*GlobalFlags) *cobra.Command { editionName := editionNameOrDefault() clientIDHash := dwsevent.ClientIDHash(clientID) workDir := eventWorkDir(configDir, editionName, dwsevent.SourceKindAppStream, clientIDHash) - if err := eventStopBus(busctl.StopConfig{WorkDir: workDir}); err != nil { + ipcEndpoint := defaultIPCEndpoint(workDir, editionName, dwsevent.SourceKindAppStream, clientIDHash) + if err := eventStopBus(busctl.StopConfig{WorkDir: workDir, IPCEndpoint: ipcEndpoint}); err != nil { if errors.Is(err, busctl.ErrNotRunning) { fmt.Fprintln(c.OutOrStdout(), "bus is not running") return nil diff --git a/internal/app/event_personal_command.go b/internal/app/event_personal_command.go index b2b1dc03..20ff6207 100644 --- a/internal/app/event_personal_command.go +++ b/internal/app/event_personal_command.go @@ -1220,7 +1220,7 @@ func runPersonalEventStop(c *cobra.Command, opts personalStopOptions) error { } busState := "personal bus stopped" - if err := personalStopBus(busctl.StopConfig{WorkDir: workDir}); err != nil { + if err := personalStopBus(busctl.StopConfig{WorkDir: workDir, IPCEndpoint: ipcEndpoint}); err != nil { if errors.Is(err, busctl.ErrNotRunning) { busState = "personal bus is not running" } else { diff --git a/internal/event/bus/coverage_edges_test.go b/internal/event/bus/coverage_edges_test.go index 8086e4d7..ea0f9cbb 100644 --- a/internal/event/bus/coverage_edges_test.go +++ b/internal/event/bus/coverage_edges_test.go @@ -361,13 +361,18 @@ func TestCrossPlatformCoverageDaemonMethodEdges(t *testing.T) { } return nil, net.ErrClosed }} - d := &daemon{listener: l, log: logger, hub: NewHub(1), idleStop: make(chan struct{})} + d := &daemon{listener: l, log: logger, hub: NewHub(1), idleStop: make(chan struct{}), stopReq: make(chan string, 1)} d.acceptLoop(context.Background()) d.shuttingDown.Store(true) d.acceptLoop(context.Background()) d.triggerShutdown("test") - if !l.closed { - t.Fatal("trigger did not close listener") + select { + case reason := <-d.stopReq: + if reason != "test" { + t.Fatalf("stop reason = %q", reason) + } + default: + t.Fatal("trigger did not notify lifecycle loop") } d = &daemon{listener: &scriptedListener{accept: func() (net.Conn, error) { return nil, net.ErrClosed }}, log: logger, hub: NewHub(1), idleStop: make(chan struct{})} diff --git a/internal/event/bus/daemon.go b/internal/event/bus/daemon.go index 7fac9aff..0c43d908 100644 --- a/internal/event/bus/daemon.go +++ b/internal/event/bus/daemon.go @@ -186,6 +186,7 @@ func Run(ctx context.Context, cfg Config) error { dedup: dd, started: time.Now().UTC(), idleStop: make(chan struct{}), + stopReq: make(chan string, 1), } // 4. Signal ready BEFORE accepting consumers (avoids a slow-fork @@ -263,6 +264,9 @@ func Run(ctx context.Context, cfg Config) error { } case <-d.idleStop: log.Info("bus: idle timeout reached, shutting down") + case reason := <-d.stopReq: + shutdownReason = reason + log.Info("bus: shutdown requested via IPC", "reason", reason) } // 7. Graceful shutdown — cancel runCtx first so all background @@ -292,6 +296,7 @@ type daemon struct { shutdownMu sync.Mutex shuttingDown atomic.Bool idleStop chan struct{} + stopReq chan string credentialHandoffMu sync.Mutex terminalMu sync.RWMutex @@ -749,15 +754,17 @@ func (d *daemon) idleWatch(ctx context.Context) { } } -// triggerShutdown is called from RPC handlers that want to end the bus. -// It works by closing the listener (which unblocks Run's select via the -// source error path, indirectly). For v1 a full ctx-cancellation hook is -// out of scope; busctl/stop also sends SIGTERM which is the authoritative -// shutdown path. +// triggerShutdown is called from RPC handlers that want to end the bus. The +// buffered request wakes Run's lifecycle select, which then cancels the cloud +// source and performs the same graceful shutdown used for signals and idle +// expiry. It is intentionally non-blocking so repeated stop requests cannot +// strand connection handlers. func (d *daemon) triggerShutdown(reason string) { d.log.Info("bus: shutdown triggered via IPC", "reason", reason) - _ = d.listener.Close() // unblocks Accept(), but doesn't kill Source - // Best-effort: a future version wires a context.CancelFunc here. + select { + case d.stopReq <- reason: + default: + } } // shutdown performs the graceful tear-down sequence: diff --git a/internal/event/bus/daemon_test.go b/internal/event/bus/daemon_test.go index 49a46d20..d99e4aa7 100644 --- a/internal/event/bus/daemon_test.go +++ b/internal/event/bus/daemon_test.go @@ -141,6 +141,67 @@ func TestDaemon_RunStartsAndShutsDownCleanly(t *testing.T) { } } +func TestCrossPlatformCoverageDaemonIPCStopShutsDownCleanly(t *testing.T) { + workDir := shortTempDir(t) + identityHash := dwsevent.IdentityHash(workDir) + endpoint := dwsevent.IPCEndpoint(workDir, "open", dwsevent.SourceKindAppStream, identityHash) + runDone := make(chan error, 1) + go func() { + runDone <- Run(context.Background(), Config{ + WorkDir: workDir, + IPCEndpoint: endpoint, + ClientID: "ding_ipc_stop_test", + Edition: "open", + SourceKind: dwsevent.SourceKindAppStream, + IdentityHash: identityHash, + Source: &fakeSource{}, + }) + }() + + var conn net.Conn + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + var err error + conn, err = transport.Dial(endpoint) + if err == nil { + break + } + time.Sleep(10 * time.Millisecond) + } + if conn == nil { + t.Fatal("bus IPC endpoint did not become ready") + } + w := transport.NewWriter(conn) + r := transport.NewReader(conn) + if err := w.WriteJSON(transport.Hello{ + Type: transport.FrameTypeHello, + ConsumerPID: os.Getpid(), + Role: transport.HelloRoleStop, + }); err != nil { + t.Fatalf("write stop hello: %v", err) + } + var bye transport.Bye + if err := r.ReadJSON(&bye); err != nil { + t.Fatalf("read stop response: %v", err) + } + _ = conn.Close() + if bye.Type != transport.FrameTypeBye || bye.Reason != "stop_request" { + t.Fatalf("stop response = %#v", bye) + } + + select { + case err := <-runDone: + if err != nil { + t.Fatalf("Run returned after IPC stop: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("Run did not return after IPC stop") + } + if pid := ReadHolderPID(filepath.Join(workDir, LockFileName)); pid != 0 { + t.Fatalf("bus lock retained pid %d after IPC stop", pid) + } +} + func TestDaemon_ConsumerReceivesEvents(t *testing.T) { skipOnWindows(t, "uses Unix socket dial") workDir := shortTempDir(t) diff --git a/internal/event/busctl/discover.go b/internal/event/busctl/discover.go index 2bbbb863..56fd1c64 100644 --- a/internal/event/busctl/discover.go +++ b/internal/event/busctl/discover.go @@ -112,8 +112,9 @@ func Discover(cfg DiscoverConfig) (net.Conn, error) { return nil, fmt.Errorf("busctl: mkdir workdir: %w", err) } _, spawnErr := cfg.Spawn(SpawnConfig{ - ClientID: cfg.ClientID, - ExtraArgs: cfg.SpawnExtraArgs, + ClientID: cfg.ClientID, + IPCEndpoint: cfg.IPCEndpoint, + ExtraArgs: cfg.SpawnExtraArgs, }) if spawnErr != nil && !errors.Is(spawnErr, ErrSpawnFailed) { // Hard error (couldn't even exec the child). Stop here — no bus diff --git a/internal/event/busctl/discover_test.go b/internal/event/busctl/discover_test.go index a4b42bcb..7e2e2dbc 100644 --- a/internal/event/busctl/discover_test.go +++ b/internal/event/busctl/discover_test.go @@ -122,7 +122,10 @@ func TestDiscover_NoBus_SpawnSucceeds(t *testing.T) { closer() } }) - fakeSpawn := func(SpawnConfig) (int, error) { + fakeSpawn := func(cfg SpawnConfig) (int, error) { + if cfg.IPCEndpoint != sock { + t.Fatalf("spawn IPC endpoint = %q, want %q", cfg.IPCEndpoint, sock) + } closer = startStubBus(t, sock) return 12345, nil } diff --git a/internal/event/busctl/spawn.go b/internal/event/busctl/spawn.go index 5c3c589f..f3b89165 100644 --- a/internal/event/busctl/spawn.go +++ b/internal/event/busctl/spawn.go @@ -43,18 +43,14 @@ var ( spawnPipe = os.Pipe ) -// ErrSpawnFailed is returned when the child reports startup failure via -// the ready pipe ('E' byte). The child's exit error / log file holds the -// actual cause; this sentinel just lets the caller distinguish "ready -// pipe said no" from "ready pipe timed out / closed early". -var ErrSpawnFailed = errors.New("busctl: bus child reported startup failure on ready pipe") +// ErrSpawnFailed is returned when the child reports startup failure through +// the Unix ready pipe or exits before binding the Windows named pipe. +var ErrSpawnFailed = errors.New("busctl: bus child reported startup failure") // ErrSpawnTimeout is returned when ReadyTimeout elapses without any signal. var ErrSpawnTimeout = errors.New("busctl: bus child did not signal readiness within deadline") -// SpawnConfig describes one spawn attempt. ClientID is the only field -// inspected by the child; the rest govern process attributes the parent -// applies before exec. +// SpawnConfig describes one spawn attempt. type SpawnConfig struct { // ExecPath is the dws binary to exec. Default os.Executable(). ExecPath string @@ -62,18 +58,23 @@ type SpawnConfig struct { // ClientID is passed as `--client-id` to `dws event _bus`. Required. ClientID string + // IPCEndpoint is the bus endpoint the child binds. Windows uses the + // existing named pipe as its readiness handshake because os/exec does not + // support ExtraFiles there. Unix keeps the inherited ready-pipe protocol. + IPCEndpoint string + // ExtraArgs are appended after `--client-id`. Empty for normal use; tests // pass `--extra-flag-for-test` etc. ExtraArgs []string - // Env to pass to the child. Defaults to os.Environ(). The ReadyFDEnv - // entry is appended automatically. + // Env to pass to the child. Defaults to os.Environ(). Unix appends the + // ReadyFDEnv entry automatically; Windows removes any inherited copy. Env []string } -// Spawn forks a detached `dws event _bus --client-id ` child process and -// waits for it to signal readiness via the ready pipe. Returns the child's -// PID on success — the caller can then dial the bus IPC endpoint. +// spawnWithReadyPipe is the Unix implementation behind Spawn. It forks a +// detached `dws event _bus --client-id ` child and waits for the inherited +// ready pipe. Windows has a named-pipe implementation in spawn_windows.go. // // stdio detach (plan invariant #7): // - cmd.Stdout / cmd.Stderr set to nil so the child's own writes don't @@ -89,7 +90,7 @@ type SpawnConfig struct { // Parent (this function): // - Holds the read end open until either 1 byte is read or ReadyTimeout // - Returns ErrSpawnFailed for 'E', ErrSpawnTimeout otherwise -func Spawn(cfg SpawnConfig) (pid int, err error) { +func spawnWithReadyPipe(cfg SpawnConfig) (pid int, err error) { if cfg.ClientID == "" { return 0, errors.New("busctl: SpawnConfig.ClientID is required") } diff --git a/internal/event/busctl/spawn_test.go b/internal/event/busctl/spawn_test.go index 42a1c532..dc931700 100644 --- a/internal/event/busctl/spawn_test.go +++ b/internal/event/busctl/spawn_test.go @@ -22,6 +22,8 @@ import ( "strings" "testing" "time" + + "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport" ) // Test child mode: when DWS_BUSCTL_TEST_CHILD is set, this test binary @@ -35,7 +37,8 @@ import ( // env-marker pattern is what Go's own os/exec tests use and stays // confined to this file. const ( - childEnvMarker = "DWS_BUSCTL_TEST_CHILD" + childEnvMarker = "DWS_BUSCTL_TEST_CHILD" + childEndpointEnv = "DWS_BUSCTL_TEST_ENDPOINT" // values: // "ready" — write 'R' then sleep 30s (parent should see ready) // "fail" — write 'E' then exit (parent should see ErrSpawnFailed) @@ -70,6 +73,26 @@ func TestMain(m *testing.M) { writeReady('R') time.Sleep(30 * time.Second) os.Exit(0) + case "windows-ready": + listener, err := transport.Listen(os.Getenv(childEndpointEnv)) + if err != nil { + os.Exit(3) + } + defer listener.Close() + conn, err := listener.Accept() + if err != nil { + os.Exit(4) + } + _ = conn.Close() + time.Sleep(30 * time.Second) + os.Exit(0) + case "windows-fail": + os.Exit(5) + case "windows-exit": + os.Exit(0) + case "windows-stall": + time.Sleep(30 * time.Second) + os.Exit(0) } os.Exit(m.Run()) } diff --git a/internal/event/busctl/spawn_unix.go b/internal/event/busctl/spawn_unix.go index b568c524..d882808c 100644 --- a/internal/event/busctl/spawn_unix.go +++ b/internal/event/busctl/spawn_unix.go @@ -20,6 +20,13 @@ import ( "syscall" ) +// Spawn starts the detached bus and waits for its inherited ready pipe. +// ExtraFiles is intentionally confined to Unix: Go does not support it on +// Windows. +func Spawn(cfg SpawnConfig) (pid int, err error) { + return spawnWithReadyPipe(cfg) +} + // applyDetach configures the child to live past parent death and not share // the parent's controlling terminal. Setsid puts the child in a new // session, so SIGHUP on the parent's controlling tty (e.g. SSH disconnect) diff --git a/internal/event/busctl/spawn_windows.go b/internal/event/busctl/spawn_windows.go index ee0174b1..1cdf6958 100644 --- a/internal/event/busctl/spawn_windows.go +++ b/internal/event/busctl/spawn_windows.go @@ -16,14 +16,101 @@ package busctl import ( + "fmt" + "os" "os/exec" + "strings" "syscall" + "time" + + "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport" ) // CREATE_NEW_PROCESS_GROUP (0x00000200) prevents the child from receiving // the parent's Ctrl+C signal, similar in spirit to Setsid on Unix. const createNewProcessGroup = 0x00000200 +var ( + spawnWindowsDial = transport.Dial + spawnWindowsPollInterval = 25 * time.Millisecond +) + +// Spawn starts a detached Windows bus without cmd.ExtraFiles (unsupported by +// Go on Windows). Readiness is confirmed by dialing the bus's existing named +// pipe. The child is reaped in the background after readiness, while an early +// process exit is surfaced as ErrSpawnFailed. +func Spawn(cfg SpawnConfig) (pid int, err error) { + if strings.TrimSpace(cfg.ClientID) == "" { + return 0, fmt.Errorf("busctl: SpawnConfig.ClientID is required") + } + if strings.TrimSpace(cfg.IPCEndpoint) == "" { + return 0, fmt.Errorf("busctl: SpawnConfig.IPCEndpoint is required on Windows") + } + if cfg.ExecPath == "" { + execPath, err := spawnExecutable() + if err != nil { + return 0, fmt.Errorf("busctl: locate executable: %w", err) + } + cfg.ExecPath = execPath + } + if cfg.Env == nil { + cfg.Env = os.Environ() + } + + args := append([]string{"event", "_bus", "--client-id", cfg.ClientID}, cfg.ExtraArgs...) + cmd := exec.Command(cfg.ExecPath, args...) + cmd.Env = withoutReadyFDEnv(cfg.Env) + cmd.Stdin = nil + cmd.Stdout = nil + cmd.Stderr = nil + applyDetach(cmd) + if err := cmd.Start(); err != nil { + return 0, fmt.Errorf("busctl: start %s: %w", cfg.ExecPath, err) + } + pid = cmd.Process.Pid + + waitDone := make(chan error, 1) + go func() { waitDone <- cmd.Wait() }() + timer := time.NewTimer(ReadyTimeout) + defer timer.Stop() + ticker := time.NewTicker(spawnWindowsPollInterval) + defer ticker.Stop() + + for { + if conn, dialErr := spawnWindowsDial(cfg.IPCEndpoint); dialErr == nil { + _ = conn.Close() + return pid, nil + } + select { + case waitErr := <-waitDone: + if waitErr == nil { + return pid, fmt.Errorf("%w: child exited before binding named pipe", ErrSpawnFailed) + } + return pid, fmt.Errorf("%w: %v", ErrSpawnFailed, waitErr) + case <-timer.C: + _ = cmd.Process.Kill() + select { + case <-waitDone: + case <-time.After(time.Second): + } + return pid, ErrSpawnTimeout + case <-ticker.C: + } + } +} + +func withoutReadyFDEnv(env []string) []string { + prefix := strings.ToUpper(ReadyFDEnv) + "=" + out := make([]string, 0, len(env)) + for _, entry := range env { + if strings.HasPrefix(strings.ToUpper(entry), prefix) { + continue + } + out = append(out, entry) + } + return out +} + func applyDetach(cmd *exec.Cmd) { cmd.SysProcAttr = &syscall.SysProcAttr{ CreationFlags: createNewProcessGroup, diff --git a/internal/event/busctl/spawn_windows_test.go b/internal/event/busctl/spawn_windows_test.go new file mode 100644 index 00000000..1eeafc4a --- /dev/null +++ b/internal/event/busctl/spawn_windows_test.go @@ -0,0 +1,108 @@ +// Copyright 2026 Alibaba Group +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build windows + +package busctl + +import ( + "errors" + "fmt" + "os" + "strings" + "testing" + "time" + + "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/testseam" +) + +func TestCrossPlatformCoverageWindowsSpawnUsesNamedPipeReadiness(t *testing.T) { + endpoint := fmt.Sprintf(`\\.\pipe\dws-event-spawn-test-%d-%d`, os.Getpid(), time.Now().UnixNano()) + pid, err := spawnWithMarker(t, "windows-ready", func(cfg *SpawnConfig) { + cfg.IPCEndpoint = endpoint + cfg.Env = append(cfg.Env, childEndpointEnv+"="+endpoint) + }) + if err != nil { + t.Fatalf("Spawn() = %v", err) + } + if pid <= 0 { + t.Fatalf("Spawn() pid = %d", pid) + } + proc, err := os.FindProcess(pid) + if err != nil { + t.Fatal(err) + } + _ = proc.Kill() +} + +func TestCrossPlatformCoverageWindowsSpawnReportsEarlyExit(t *testing.T) { + endpoint := fmt.Sprintf(`\\.\pipe\dws-event-spawn-fail-%d-%d`, os.Getpid(), time.Now().UnixNano()) + _, err := spawnWithMarker(t, "windows-fail", func(cfg *SpawnConfig) { + cfg.IPCEndpoint = endpoint + }) + if !errors.Is(err, ErrSpawnFailed) { + t.Fatalf("Spawn() error = %v, want ErrSpawnFailed", err) + } +} + +func TestCrossPlatformCoverageWindowsSpawnReportsCleanEarlyExit(t *testing.T) { + endpoint := fmt.Sprintf(`\\.\pipe\dws-event-spawn-exit-%d-%d`, os.Getpid(), time.Now().UnixNano()) + _, err := spawnWithMarker(t, "windows-exit", func(cfg *SpawnConfig) { + cfg.IPCEndpoint = endpoint + }) + if !errors.Is(err, ErrSpawnFailed) { + t.Fatalf("Spawn() error = %v, want ErrSpawnFailed", err) + } +} + +func TestCrossPlatformCoverageWindowsSpawnValidationAndStartErrors(t *testing.T) { + endpoint := fmt.Sprintf(`\\.\pipe\dws-event-spawn-errors-%d-%d`, os.Getpid(), time.Now().UnixNano()) + if _, err := Spawn(SpawnConfig{ClientID: "client"}); err == nil { + t.Fatal("missing endpoint unexpectedly succeeded") + } + testseam.Swap(t, &spawnExecutable, func() (string, error) { + return "", errors.New("executable unavailable") + }) + if _, err := Spawn(SpawnConfig{ClientID: "client", IPCEndpoint: endpoint}); err == nil { + t.Fatal("executable lookup unexpectedly succeeded") + } + if _, err := Spawn(SpawnConfig{ + ClientID: "client", + IPCEndpoint: endpoint, + ExecPath: `C:\\definitely-missing-dws.exe`, + }); err == nil { + t.Fatal("missing executable unexpectedly started") + } +} + +func TestCrossPlatformCoverageWindowsSpawnReadyTimeoutTerminatesChild(t *testing.T) { + endpoint := fmt.Sprintf(`\\.\pipe\dws-event-spawn-stall-%d-%d`, os.Getpid(), time.Now().UnixNano()) + testseam.Swap(t, &ReadyTimeout, 100*time.Millisecond) + _, err := spawnWithMarker(t, "windows-stall", func(cfg *SpawnConfig) { + cfg.IPCEndpoint = endpoint + }) + if !errors.Is(err, ErrSpawnTimeout) { + t.Fatalf("Spawn() error = %v, want ErrSpawnTimeout", err) + } +} + +func TestCrossPlatformCoverageWindowsSpawnRemovesInheritedReadyFD(t *testing.T) { + env := withoutReadyFDEnv([]string{ + "PATH=C:\\Windows", + ReadyFDEnv + "=3", + strings.ToLower(ReadyFDEnv) + "=4", + }) + if len(env) != 1 || !strings.HasPrefix(env[0], "PATH=") { + t.Fatalf("withoutReadyFDEnv() = %#v", env) + } +} diff --git a/internal/event/busctl/stop.go b/internal/event/busctl/stop.go index 8b9bbdb1..c68ebfcd 100644 --- a/internal/event/busctl/stop.go +++ b/internal/event/busctl/stop.go @@ -17,15 +17,17 @@ import ( "errors" "fmt" "os" + "strings" "time" "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/bus" "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/process" + "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport" ) -// DefaultStopTimeout is the wall-clock budget Stop waits for the bus to -// exit after the signal is sent. 5s covers the bus's own graceful tear-down -// (broadcast Bye → consumer goroutines drain → cleanup) with margin. +// DefaultStopTimeout is the wall-clock budget Stop waits for each graceful or +// fallback termination phase. 5s covers the bus's own tear-down (broadcast +// Bye → consumer goroutines drain → cleanup) with margin. const DefaultStopTimeout = 5 * time.Second // ErrNotRunning indicates bus.lock either does not exist or its recorded @@ -38,15 +40,19 @@ var ( stopAlive = process.Alive stopFindProcess = os.FindProcess stopSignalProcess = func(proc *os.Process, signal os.Signal) error { return proc.Signal(signal) } + stopRequest = requestBusStop ) // StopConfig identifies the target bus and tunes timing. type StopConfig struct { // WorkDir holds bus.lock; Stop reads the PID from there. WorkDir string - // Timeout is the total wall-clock budget for graceful exit. After this, - // Stop returns an error; it does NOT escalate to SIGKILL — leave that - // to the operator. + // IPCEndpoint enables the cross-platform graceful stop RPC. Callers that + // know the bus identity should always provide it. Empty preserves the + // legacy signal-only fallback for compatibility and focused tests. + IPCEndpoint string + // Timeout is the wall-clock budget for graceful exit and, if required, + // the subsequent platform termination fallback. Timeout time.Duration } @@ -54,13 +60,12 @@ type StopConfig struct { // for the process to actually die. Returns ErrNotRunning if no bus is // running for that work dir. // -// Implementation note: on Unix we send SIGTERM. The bus daemon's Run loop -// watches its parent ctx for cancellation; the cobra `event _bus` command -// wires signal.NotifyContext so SIGTERM triggers ctx.Done() → graceful -// shutdown path. On Windows we use os.Process.Signal(os.Interrupt) which -// the Go runtime maps to TerminateProcess for processes outside our -// console group; for v1 that's acceptable (Windows graceful shutdown is -// future work — plan §16 v2). +// Stop first asks the bus to shut down through its owner-only IPC endpoint. +// This is the normal path on every platform and lets the daemon cancel its +// cloud source, notify consumers, and release its lock. If the endpoint is +// unavailable (for example an older bus) or graceful shutdown times out, the +// platform signal is used as a compatibility fallback: SIGTERM on Unix and +// TerminateProcess via os.Kill on Windows. func Stop(cfg StopConfig) error { if cfg.WorkDir == "" { return errors.New("busctl: StopConfig.WorkDir is required") @@ -81,6 +86,12 @@ func Stop(cfg StopConfig) error { if err != nil { return fmt.Errorf("busctl: find process %d: %w", pid, err) } + if strings.TrimSpace(cfg.IPCEndpoint) != "" { + if err := stopRequest(cfg.IPCEndpoint); err == nil && waitForBusExit(pid, cfg.Timeout) { + return nil + } + } + if err := stopSignalProcess(proc, stopSignal()); err != nil { // On many Unix platforms Signal returns "process already finished" // when the bus has just exited on its own — treat that as success. @@ -90,13 +101,45 @@ func Stop(cfg StopConfig) error { return fmt.Errorf("busctl: signal bus pid=%d: %w", pid, err) } - // Poll for actual exit. - deadline := time.Now().Add(cfg.Timeout) - for time.Now().Before(deadline) { - if !stopAlive(pid) { - return nil - } - time.Sleep(50 * time.Millisecond) + if waitForBusExit(pid, cfg.Timeout) { + return nil } return fmt.Errorf("busctl: bus pid=%d did not exit within %s", pid, cfg.Timeout) } + +func requestBusStop(endpoint string) error { + conn, err := transport.Dial(endpoint) + if err != nil { + return fmt.Errorf("busctl: dial bus for stop: %w", err) + } + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(DefaultStatusRPCTimeout)) + w := transport.NewWriter(conn) + r := transport.NewReader(conn) + if err := w.WriteJSON(transport.Hello{ + Type: transport.FrameTypeHello, + ConsumerPID: os.Getpid(), + Role: transport.HelloRoleStop, + }); err != nil { + return fmt.Errorf("busctl: write stop hello: %w", err) + } + var bye transport.Bye + if err := r.ReadJSON(&bye); err != nil { + return fmt.Errorf("busctl: read stop response: %w", err) + } + if bye.Type != transport.FrameTypeBye || bye.Reason != "stop_request" { + return fmt.Errorf("busctl: unexpected stop response type=%q reason=%q", bye.Type, bye.Reason) + } + return nil +} + +func waitForBusExit(pid int, timeout time.Duration) bool { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if !stopAlive(pid) { + return true + } + time.Sleep(50 * time.Millisecond) + } + return !stopAlive(pid) +} diff --git a/internal/event/busctl/stop_test.go b/internal/event/busctl/stop_test.go index e4cd99de..2a80d01e 100644 --- a/internal/event/busctl/stop_test.go +++ b/internal/event/busctl/stop_test.go @@ -23,7 +23,10 @@ import ( "testing" "time" + dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event" "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/bus" + "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport" + "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/testseam" ) func TestStop_NotRunningWhenLockMissing(t *testing.T) { @@ -46,6 +49,87 @@ func TestStop_NotRunningWhenPIDDead(t *testing.T) { } } +func TestCrossPlatformCoverageStopPrefersGracefulIPC(t *testing.T) { + const pid = 4242 + stopped := false + signals := 0 + testseam.Swap(t, &stopReadHolderPID, func(string) int { return pid }) + testseam.Swap(t, &stopAlive, func(int) bool { return !stopped }) + proc, err := os.FindProcess(os.Getpid()) + if err != nil { + t.Fatal(err) + } + testseam.Swap(t, &stopFindProcess, func(int) (*os.Process, error) { return proc, nil }) + testseam.Swap(t, &stopRequest, func(endpoint string) error { + if endpoint != "test-endpoint" { + t.Fatalf("stop endpoint = %q", endpoint) + } + stopped = true + return nil + }) + testseam.Swap(t, &stopSignalProcess, func(*os.Process, os.Signal) error { + signals++ + return nil + }) + + if err := Stop(StopConfig{ + WorkDir: "test-workdir", + IPCEndpoint: "test-endpoint", + Timeout: time.Second, + }); err != nil { + t.Fatalf("Stop() = %v", err) + } + if signals != 0 { + t.Fatalf("graceful IPC stop used %d process signals", signals) + } +} + +func TestCrossPlatformCoverageRequestBusStopProtocol(t *testing.T) { + dir := shortTempDir(t) + endpoint := dwsevent.IPCEndpoint( + dir, + "open", + dwsevent.SourceKindAppStream, + dwsevent.IdentityHash(dir), + ) + listener, err := transport.Listen(endpoint) + if err != nil { + t.Fatalf("Listen() = %v", err) + } + defer listener.Close() + serverDone := make(chan error, 1) + go func() { + conn, err := listener.Accept() + if err != nil { + serverDone <- err + return + } + defer conn.Close() + r := transport.NewReader(conn) + w := transport.NewWriter(conn) + var hello transport.Hello + if err := r.ReadJSON(&hello); err != nil { + serverDone <- err + return + } + if hello.Type != transport.FrameTypeHello || hello.Role != transport.HelloRoleStop { + serverDone <- errors.New("unexpected stop hello") + return + } + serverDone <- w.WriteJSON(transport.Bye{ + Type: transport.FrameTypeBye, + Reason: "stop_request", + }) + }() + + if err := requestBusStop(endpoint); err != nil { + t.Fatalf("requestBusStop() = %v", err) + } + if err := <-serverDone; err != nil { + t.Fatalf("stop protocol server = %v", err) + } +} + func TestStop_SignalsLiveProcess(t *testing.T) { skipOnWindows(t) dir := shortTempDir(t) diff --git a/internal/event/busctl/stop_windows.go b/internal/event/busctl/stop_windows.go index 2e7223f6..630ee330 100644 --- a/internal/event/busctl/stop_windows.go +++ b/internal/event/busctl/stop_windows.go @@ -17,8 +17,7 @@ package busctl import "os" -// stopSignal returns the graceful-shutdown signal for Windows. The Go -// runtime maps os.Interrupt to TerminateProcess for non-console-group -// processes — not truly graceful, but acceptable for v1 (Windows graceful -// shutdown via Ctrl+Break is in the v2 backlog, plan §16). -func stopSignal() os.Signal { return os.Interrupt } +// stopSignal is only the fallback after the graceful IPC stop path fails or +// times out. Go maps os.Kill to TerminateProcess on Windows; os.Interrupt is +// unsupported and returns syscall.EWINDOWS. +func stopSignal() os.Signal { return os.Kill } diff --git a/internal/event/busctl/stop_windows_test.go b/internal/event/busctl/stop_windows_test.go new file mode 100644 index 00000000..fcdc94aa --- /dev/null +++ b/internal/event/busctl/stop_windows_test.go @@ -0,0 +1,27 @@ +// Copyright 2026 Alibaba Group +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build windows + +package busctl + +import ( + "os" + "testing" +) + +func TestCrossPlatformCoverageWindowsStopFallbackUsesKill(t *testing.T) { + if stopSignal() != os.Kill { + t.Fatalf("stopSignal() = %v, want os.Kill", stopSignal()) + } +}