dingtalk-workspace-cli/internal/event/busctl/stop.go

// 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.

package busctl

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 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
// PID is not alive. Stop returns this as a sentinel so the caller can
// distinguish "nothing to stop" from "failed to stop".
var ErrNotRunning = errors.New("busctl: bus is not running")

// ErrOwnerUnverified indicates the PID from bus.lock is alive but no longer
// owns that lock. Stop refuses to send a process-level signal in this state
// because the operating system may have reused the stale PID.
var ErrOwnerUnverified = errors.New("busctl: bus process ownership could not be verified")

var (
	stopReadHolderPID       = bus.ReadHolderPID
	stopValidateHolderOwner = bus.ValidateHolderOwnership
	stopAlive               = process.Alive
	stopFindProcess         = os.FindProcess
	stopSignalProcess       = func(proc *os.Process, signal os.Signal) error { return proc.Signal(signal) }
	stopDial                = transport.Dial
	stopRequest             = requestBusStop
	stopWaitForBusExit      = waitForBusExit
)

// StopConfig identifies the target bus and tunes timing.
type StopConfig struct {
	// WorkDir holds bus.lock; Stop reads the PID from there.
	WorkDir string
	// 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
}

// Stop signals the bus daemon for cfg.WorkDir to exit gracefully and waits
// for the process to actually die. Returns ErrNotRunning if no bus is
// running for that work dir.
//
// 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")
	}
	if cfg.Timeout == 0 {
		cfg.Timeout = DefaultStopTimeout
	}

	pid := stopReadHolderPID(LockPath(cfg.WorkDir))
	if pid <= 0 {
		return ErrNotRunning
	}
	if !stopAlive(pid) {
		return ErrNotRunning
	}

	proc, err := stopFindProcess(pid)
	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 && stopWaitForBusExit(pid, cfg.Timeout) {
			return nil
		}
	}

	// The bus may finish shutting down immediately after the graceful wait
	// reaches its deadline (or after a failed IPC request). Recheck before
	// inspecting ownership so a completed stop is not reported as stale.
	if !stopAlive(pid) {
		return nil
	}
	owner, err := stopValidateHolderOwner(LockPath(cfg.WorkDir), pid)
	if err != nil {
		return fmt.Errorf("busctl: verify bus pid=%d ownership: %w", pid, err)
	}
	if !owner {
		// ValidateHolderOwnership observes a released lock both for a stale,
		// reused PID and for a bus that exits during the ownership probe. Only
		// the still-alive case is unsafe to signal.
		if !stopAlive(pid) {
			return nil
		}
		return fmt.Errorf("%w: pid=%d does not own %s; stale PID was not signalled", ErrOwnerUnverified, pid, LockPath(cfg.WorkDir))
	}

	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.
		if errors.Is(err, os.ErrProcessDone) {
			return nil
		}
		return fmt.Errorf("busctl: signal bus pid=%d: %w", pid, err)
	}

	if stopWaitForBusExit(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 := stopDial(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)
}