AgentServer/apps/mounter/manager.go

package main

// 挂载管理（docs/design/21-knowledge-sources.md §4.4、§4.5）：
// 平台用 sync 下发完整的登记列表；挂载进程挂上缺少的、卸载多余的、更新有变化的。
// 挂载表变化时（设备插拔）立即重新计算状态，不需要定时检查。

import (
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/exec"
	"path/filepath"
	"regexp"
	"sort"
	"sync"
	"time"

	"github.com/hanwen/go-fuse/v2/fuse"
	"golang.org/x/sys/unix"
)

var idPattern = regexp.MustCompile(`^kb_[0-9a-z]{1,64}$`)

type mountSpec struct {
	ID     string       `json:"id"`
	Kind   string       `json:"kind"`
	Local  *localSource `json:"local,omitempty"`
	Remote *remoteSpec  `json:"remote,omitempty"`
}

type mountStatus struct {
	ID      string  `json:"id"`
	State   string  `json:"state"`
	Message *string `json:"message"`
	Since   string  `json:"since"`
	// 当前的实际路径（本机目录）
	RealPath string `json:"realPath,omitempty"`
}

type mount struct {
	spec      mountSpec
	point     string
	src       *source
	server    *fuse.Server
	state     string
	message   string
	since     time.Time
	realPath  string
	remote    *remoteProc // 远程存储的 rclone 挂载
	remoteErr string      // 远程存储配置不正确
}

type manager struct {
	mu        sync.Mutex
	sys       system
	roots     []string
	mntDir    string
	stateFile string
	mounts    map[string]*mount
	lastMiss  time.Time
	// mountFn 便于测试替换
	mountFn func(point, name string, src *source) (*fuse.Server, error)
	// 远程存储：rclone 的内部挂载点与缓存目录
	remoteDir string
	cacheDir  string
	runner    *rcloneRunner
}

func newManager(sys system, roots []string, mntDir, stateFile string) *manager {
	return &manager{sys: sys, roots: roots, mntDir: mntDir, stateFile: stateFile, mounts: map[string]*mount{}, mountFn: mountFS}
}

// withRemote 启用远程存储（rclone 程序、内部挂载点目录、缓存目录）
func (m *manager) withRemote(runner *rcloneRunner, remoteDir, cacheDir string) *manager {
	m.runner, m.remoteDir, m.cacheDir = runner, remoteDir, cacheDir
	return m
}

func (m *manager) setState(mt *mount, state, msg string) {
	if mt.state != state || mt.message != msg {
		if mt.state != "" {
			log.Printf("%s：%s → %s %s", mt.spec.ID, mt.state, state, msg)
		}
		mt.state, mt.message, mt.since = state, msg, time.Now()
	}
}

// resolve 按登记找到源目录并检查，返回真实路径与设备号、inode 号
func (m *manager) resolve(spec mountSpec) (string, uint64, uint64, *checkError) {
	if spec.Kind != "local" || spec.Local == nil {
		return "", 0, 0, &checkError{"mount_failed", "暂不支持这种资料库类型"}
	}
	p, cerr := locate(m.sys, *spec.Local)
	if cerr != nil {
		return "", 0, 0, cerr
	}
	real, cerr := checkSource(p, m.roots)
	if cerr != nil {
		return "", 0, 0, cerr
	}
	how := unix.OpenHow{Flags: unix.O_PATH | unix.O_DIRECTORY | unix.O_CLOEXEC, Resolve: unix.RESOLVE_NO_SYMLINKS | unix.RESOLVE_NO_MAGICLINKS}
	fd, err := unix.Openat2(unix.AT_FDCWD, real, &how)
	if err != nil {
		return "", 0, 0, &checkError{"path_missing", "无法打开目录：" + real}
	}
	defer unix.Close(fd)
	var st unix.Stat_t
	if err := unix.Fstat(fd, &st); err != nil {
		return "", 0, 0, &checkError{"path_missing", "无法读取目录：" + real}
	}
	return real, st.Dev, st.Ino, nil
}

// resolveRemote 远程存储：rclone 已挂载时，源目录为内部挂载点；状态来自最近一次连接检查
func (m *manager) resolveRemote(mt *mount) (string, uint64, uint64, string, string) {
	if mt.remoteErr != "" {
		return "", 0, 0, "mount_failed", mt.remoteErr
	}
	if mt.remote == nil {
		return "", 0, 0, "mount_failed", "没有配置 rclone"
	}
	mounted, state, msg := mt.remote.status()
	if !mounted || !mt.remote.isMounted() {
		if state == "ok" {
			state, msg = "pending", "正在连接"
		}
		return "", 0, 0, state, msg
	}
	var st unix.Stat_t
	if err := unix.Stat(mt.remote.point, &st); err != nil {
		return "", 0, 0, "unreachable", "无法访问 rclone 挂载：" + err.Error()
	}
	return mt.remote.point, st.Dev, st.Ino, state, msg
}

// refreshLocked 重新计算一个挂载的状态
func (m *manager) refreshLocked(mt *mount) {
	if mt.spec.Kind == "remote" {
		real, dev, ino, state, msg := m.resolveRemote(mt)
		if real == "" {
			mt.src.close()
		} else {
			mt.src.set(real, dev, ino)
		}
		mt.realPath = ""
		if mt.server == nil && state == "ok" {
			state, msg = "mount_failed", "挂载失败"
		}
		m.setState(mt, state, msg)
		return
	}
	real, dev, ino, cerr := m.resolve(mt.spec)
	if cerr != nil {
		mt.src.close()
		mt.realPath = ""
		m.setState(mt, cerr.State, cerr.Message)
		return
	}
	mt.src.set(real, dev, ino)
	mt.realPath = real
	if mt.server == nil {
		m.setState(mt, "mount_failed", "挂载失败")
		return
	}
	m.setState(mt, "ok", "")
}

func (m *manager) refreshAll() {
	m.mu.Lock()
	defer m.mu.Unlock()
	for _, mt := range m.mounts {
		m.refreshLocked(mt)
	}
}

// onMiss 资料库不可用时有人访问：可能是设备刚接上（如自动挂载），限速重新检查
func (m *manager) onMiss() {
	m.mu.Lock()
	if time.Since(m.lastMiss) < 2*time.Second {
		m.mu.Unlock()
		return
	}
	m.lastMiss = time.Now()
	m.mu.Unlock()
	go m.refreshAll()
}

func (m *manager) mountLocked(mt *mount) {
	if mt.server != nil {
		return
	}
	if err := os.MkdirAll(mt.point, 0o755); err != nil {
		log.Printf("%s：创建挂载点失败：%v", mt.spec.ID, err)
		return
	}
	srv, err := m.mountFn(mt.point, "kb:"+mt.spec.ID, mt.src)
	if err != nil {
		log.Printf("%s：挂载失败：%v", mt.spec.ID, err)
		return
	}
	mt.server = srv
}

func (m *manager) unmountLocked(mt *mount) {
	if mt.server != nil {
		if err := mt.server.Unmount(); err != nil {
			// 有文件正被打开：延迟卸载
			_ = exec.Command("fusermount3", "-uz", mt.point).Run()
		}
		mt.server = nil
	}
	mt.src.close()
	_ = os.Remove(mt.point)
	if mt.remote != nil {
		mt.remote.shutdown()
		mt.remote = nil
	}
}

// setupRemoteLocked 远程存储：配置变化时重启 rclone，返回新启动的进程
func (m *manager) setupRemoteLocked(mt *mount) *remoteProc {
	mt.remoteErr = ""
	if m.runner == nil {
		mt.remoteErr = "挂载服务没有配置 rclone"
		return nil
	}
	env, err := buildRemote(*mt.spec.Remote, m.runner.bin)
	if err != nil {
		mt.remoteErr = err.Error()
		if mt.remote != nil {
			mt.remote.shutdown()
			mt.remote = nil
		}
		return nil
	}
	key := remoteKey(mt.spec.Remote)
	if mt.remote != nil && mt.remote.key == key {
		return nil
	}
	if mt.remote != nil {
		log.Printf("%s：远程存储配置有变化，重新连接", mt.spec.ID)
		mt.remote.shutdown()
	}
	mt.remote = newRemoteProc(mt.spec.ID, m.remoteDir, m.cacheDir, m.runner, m.sys, env, key, func() { go m.refreshAll() })
	mt.remote.start()
	return mt.remote
}

// sync 按完整列表挂载、卸载、更新
func (m *manager) sync(specs []mountSpec, save bool) ([]mountStatus, error) {
	want := map[string]mountSpec{}
	for _, s := range specs {
		if !idPattern.MatchString(s.ID) {
			return nil, fmt.Errorf("资料库 ID 格式不正确：%q", s.ID)
		}
		if s.Kind == "local" && (s.Local == nil || !filepath.IsAbs(s.Local.Path)) {
			return nil, fmt.Errorf("%s：缺少本机目录", s.ID)
		}
		if s.Kind == "remote" && s.Remote == nil {
			return nil, fmt.Errorf("%s：缺少远程存储配置", s.ID)
		}
		if s.Kind != "local" && s.Kind != "remote" {
			return nil, fmt.Errorf("%s：不支持的类型 %s", s.ID, s.Kind)
		}
		want[s.ID] = s
	}
	m.mu.Lock()
	var started []*remoteProc
	for id, mt := range m.mounts {
		if _, ok := want[id]; !ok {
			m.unmountLocked(mt)
			delete(m.mounts, id)
			log.Printf("%s：已卸载", id)
		}
	}
	for id, s := range want {
		mt := m.mounts[id]
		if mt == nil {
			mt = &mount{spec: s, point: filepath.Join(m.mntDir, id), src: newSource()}
			mt.src.onMiss = m.onMiss
			m.mounts[id] = mt
			log.Printf("%s：登记 %s", id, describeSpec(s))
		} else {
			if mt.spec.Kind != s.Kind && mt.remote != nil {
				mt.remote.shutdown()
				mt.remote = nil
			}
			mt.spec = s
		}
		if s.Kind == "remote" {
			if p := m.setupRemoteLocked(mt); p != nil {
				started = append(started, p)
			}
		}
		m.mountLocked(mt)
		m.refreshLocked(mt)
	}
	m.mu.Unlock()
	if save {
		if err := m.saveState(specs); err != nil {
			log.Printf("保存 state.json 失败：%v", err)
		}
	}
	// 新连接的远程存储：等首次得出状态（最多 20 秒），让平台拿到准确的状态
	timeout := time.After(20 * time.Second)
	for _, p := range started {
		select {
		case <-p.ready:
		case <-timeout:
		}
	}
	if len(started) > 0 {
		m.refreshAll()
	}
	return m.status(), nil
}

// checkRemotes 状态查询时，超过 30 秒没检查过的远程存储在后台重新检查
func (m *manager) checkRemotes() {
	m.mu.Lock()
	defer m.mu.Unlock()
	for _, mt := range m.mounts {
		if mt.remote != nil {
			mt.remote.checkIfStale()
		}
	}
}

func describeSpec(s mountSpec) string {
	if s.Remote != nil {
		return fmt.Sprintf("远程存储 %s %s", s.Remote.Type, s.Remote.Params["endpoint"]+s.Remote.Params["url"]+"/"+s.Remote.Params["bucket"])
	}
	if s.Local != nil {
		if s.Local.UUID != "" {
			return fmt.Sprintf("%s（UUID %s，设备内 %s）", s.Local.Path, s.Local.UUID, s.Local.RelPath)
		}
		return s.Local.Path
	}
	return s.Kind
}

func (m *manager) status() []mountStatus {
	m.mu.Lock()
	defer m.mu.Unlock()
	out := make([]mountStatus, 0, len(m.mounts))
	for id, mt := range m.mounts {
		st := mountStatus{ID: id, State: mt.state, Since: mt.since.UTC().Format(time.RFC3339), RealPath: mt.realPath}
		if mt.message != "" {
			msg := mt.message
			st.Message = &msg
		}
		out = append(out, st)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
	return out
}

func (m *manager) unmountAll() {
	m.mu.Lock()
	defer m.mu.Unlock()
	for id, mt := range m.mounts {
		m.unmountLocked(mt)
		delete(m.mounts, id)
	}
}

type stateFile struct {
	Mounts []mountSpec `json:"mounts"`
}

func (m *manager) saveState(specs []mountSpec) error {
	b, err := json.MarshalIndent(stateFile{Mounts: specs}, "", "  ")
	if err != nil {
		return err
	}
	tmp := m.stateFile + ".tmp"
	if err := os.WriteFile(tmp, b, 0o600); err != nil {
		return err
	}
	return os.Rename(tmp, m.stateFile)
}

func (m *manager) loadState() []mountSpec {
	b, err := os.ReadFile(m.stateFile)
	if err != nil {
		return nil
	}
	var s stateFile
	if json.Unmarshal(b, &s) != nil {
		log.Printf("state.json 格式不正确，忽略")
		return nil
	}
	return s.Mounts
}

// cleanupStale 启动时清理 mnt/ 下残留的挂载（上次进程退出后失效）和多余的空目录
func (m *manager) cleanupStale() {
	for _, e := range m.sys.mountinfo() {
		if filepath.Dir(e.Point) == m.mntDir || (m.remoteDir != "" && filepath.Dir(e.Point) == m.remoteDir) {
			if err := exec.Command("fusermount3", "-uz", e.Point).Run(); err != nil {
				log.Printf("清理残留挂载 %s 失败：%v", e.Point, err)
			} else {
				log.Printf("已清理残留挂载 %s", e.Point)
			}
		}
	}
	ents, _ := os.ReadDir(m.mntDir)
	for _, e := range ents {
		if idPattern.MatchString(e.Name()) {
			_ = os.Remove(filepath.Join(m.mntDir, e.Name()))
		}
	}
}

// watchMountinfo 挂载表变化时（POLLPRI）重新计算状态
func (m *manager) watchMountinfo(stop <-chan struct{}) {
	f, err := os.Open("/proc/self/mountinfo")
	if err != nil {
		log.Printf("无法监听挂载表：%v", err)
		return
	}
	defer f.Close()
	buf := make([]byte, 64*1024)
	drain := func() {
		_, _ = f.Seek(0, 0)
		for {
			n, err := f.Read(buf)
			if n == 0 || err != nil {
				return
			}
		}
	}
	drain()
	fds := []unix.PollFd{{Fd: int32(f.Fd()), Events: unix.POLLPRI}}
	for {
		select {
		case <-stop:
			return
		default:
		}
		n, err := unix.Poll(fds, 1000)
		if err != nil && err != unix.EINTR {
			log.Printf("监听挂载表出错：%v", err)
			time.Sleep(time.Second)
			continue
		}
		if n <= 0 || fds[0].Revents&(unix.POLLPRI|unix.POLLERR) == 0 {
			continue
		}
		drain()
		// 设备插入时往往连续变化几次，稍等一下再统一计算
		time.Sleep(300 * time.Millisecond)
		drain()
		m.refreshAll()
	}
}