package syncer import ( "context" "encoding/json" "fmt" "sort" "strings" "time" ) // Mode 决定这一轮允许往哪个方向写。 // // 注意这里**没有** "status/dry-run" 这一档:只报告不写入是「算完不落地」, // 不是「不许算」——两者混在一个枚举里会让 status 把该有的差异也算成"无事可做", // 从而漏报真正的不一致(实现初期正是这么错的)。只读由 Apply 的 dryRun 参数表达。 type Mode string const ( // ModeSync 双向对齐:缺的补上、冲突按判据取新的一方。 ModeSync Mode = "sync" // ModePull 只往下拉(远端 → 本地)。本地独有的文件不动。 ModePull Mode = "pull" // ModePush 只往上推(本地 → 远端)。远端独有的文件不动。 ModePush Mode = "push" ) // ParseCommand 把子命令解析为 (同步模式, 是否只报告不写入)。 // // "status" / "diff" / "plan" 一律按**双向**算一遍,只是不落地——这样它报出来的 // 差异和真正 sync 会做的事完全一致,可以直接当"预演"用。 func ParseCommand(s string) (Mode, bool, error) { switch strings.ToLower(strings.TrimSpace(s)) { case "", "sync", "all": return ModeSync, false, nil case "pull", "down", "download": return ModePull, false, nil case "push", "up", "upload": return ModePush, false, nil case "status", "diff", "plan", "dry-run", "dryrun": return ModeSync, true, nil default: return "", false, fmt.Errorf("未知子命令 %q(可选 sync/pull/push/status/watch)", s) } } // allowsUpload 报告该模式是否允许写远端。 func (m Mode) allowsUpload() bool { return m == ModeSync || m == ModePush } // allowsDownload 报告该模式是否允许写本地。 func (m Mode) allowsDownload() bool { return m == ModeSync || m == ModePull } // Op 是一个同步动作。 type Op string const ( // OpNoop 两端已一致或按模式不动。 OpNoop Op = "noop" // OpUpload 本地 → 远端。 OpUpload Op = "upload" // OpDownload 远端 → 本地。 OpDownload Op = "download" ) // Action 是计划里的一条动作。 type Action struct { Op Op `json:"op"` Path string `json:"path"` Category string `json:"category"` Reason string `json:"reason"` // Conflict 表示两端都有且内容不同(真正的冲突,而不是单边缺失)。 Conflict bool `json:"conflict"` Local *FileState `json:"local,omitempty"` Remote *FileState `json:"remote,omitempty"` // Applied / Err 由 Apply 回填。 Applied bool `json:"applied"` Err string `json:"err,omitempty"` } // Plan 是一次对账的结果。 type Plan struct { Mode Mode Actions []Action // Bucket / Writer 随计划带上,方便输出与排障时一眼看清"对的是哪个桶、会用哪条写入通道"。 Bucket string Writer string LocalFiles int RemoteFiles int } // Uploads 返回所有上传动作。 func (p *Plan) Uploads() []Action { return p.filter(OpUpload) } // Downloads 返回所有下载动作。 func (p *Plan) Downloads() []Action { return p.filter(OpDownload) } // Conflicts 返回所有真冲突(两端都有、内容不同、需要裁决)。 func (p *Plan) Conflicts() []Action { var out []Action for _, a := range p.Actions { if a.Conflict { out = append(out, a) } } return out } // Changes 报告计划里是否有实际改动。 func (p *Plan) Changes() int { return len(p.Uploads()) + len(p.Downloads()) } func (p *Plan) filter(op Op) []Action { var out []Action for _, a := range p.Actions { if a.Op == op { out = append(out, a) } } return out } // Result 是一次执行的统计。 type Result struct { Mode Mode Uploaded int Downloaded int Skipped int Failed int Actions []Action Errors []string } // Config 是同步配置。字段与 config.json 的 "sync" 段一一对应。 type Config struct { // Bucket 桶 ID,形如 "a3216/wb2api-storage"。 Bucket string `json:"bucket"` // Token 具写权限的 HF token。为空时回落到 TokenFile。 Token string `json:"token"` // TokenFile 存放 token 的文件(token 不进 config.json 时用)。 TokenFile string `json:"token_file"` // Endpoint HF 端点,默认 https://huggingface.co。 Endpoint string `json:"endpoint"` // AuthDir 本地凭证目录,与网关 auth_dir 同值。 AuthDir string `json:"auth_dir"` // DataDir 本地数据目录(state_file 所在目录)。 DataDir string `json:"data_dir"` // IncludeData 除 auths/ 之外还要同步的 data/ 顶层文件名白名单。 // 默认空 = 只同步凭证。理由见 docs:state.json 是两个进程各自的运行态, // 互相搬运会把对端的冷却/熔断状态带进来。 IncludeData []string `json:"include_data"` // RelayURL 中转网关地址(如 https://cli.3216gemini.dpdns.org),relay 写入通道用。 RelayURL string `json:"relay_url"` // RelayKey 中转网关的 api_key(Bearer)。为空且配置了桶时可自动从桶的 // config.json 读取。 RelayKey string `json:"relay_key"` // Writer 写入通道:auto(默认)/ hf / relay。 Writer string `json:"writer"` // HFBin hf 可执行文件名/路径,默认 "hf"。 HFBin string `json:"hf_bin"` // RestartSpace 形如 "a3216/gcli2api"。配置后,每当本轮向 auths/ 推了新凭证 // 就请求重启该 Space——Space 上的网关只在启动时扫凭证目录,不重启就加载不到新账号。 // 空 = 从不重启(省下 30~60s 冷启动,代价是新账号要等下次 Space 重启才生效)。 RestartSpace string `json:"restart_space"` // CABundle 额外的 CA 证书包(PEM)路径。 // // 本机跑抓包代理(DevSidecar / Reqable / ProxyPin 之类)时会出现一种很别扭的 // 不对称:Go 在 Windows 上吃系统信任库,所以**读**同步一切正常;但写通道用的 // hf CLI 底层是 httpx,只认 certifi,会把代理的中间人证书判成不可信而失败。 // 把代理根证书导出成 PEM 指到这里,就会同时喂给 Go 客户端(追加信任,不替换 // 系统库)与 hf 子进程(SSL_CERT_FILE / REQUESTS_CA_BUNDLE / CURL_CA_BUNDLE)。 CABundle string `json:"ca_bundle"` // Timeout 单次网络操作超时,默认 60s。 Timeout string `json:"timeout"` // InsecureSkipVerify 关闭 TLS 校验(仅本地抓包代理排障用,默认 false)。 InsecureSkipVerify bool `json:"insecure_skip_verify"` } // Normalize 补默认值。 func (c *Config) Normalize() { if strings.TrimSpace(c.Endpoint) == "" { c.Endpoint = DefaultEndpoint } if strings.TrimSpace(c.Writer) == "" { c.Writer = "auto" } if strings.TrimSpace(c.HFBin) == "" { c.HFBin = "hf" } for i, d := range c.IncludeData { c.IncludeData[i] = strings.TrimSpace(d) } c.Bucket = strings.Trim(strings.TrimSpace(c.Bucket), "/") c.Endpoint = strings.TrimRight(strings.TrimSpace(c.Endpoint), "/") c.RelayURL = strings.TrimRight(strings.TrimSpace(c.RelayURL), "/") } // timeout 解析 Timeout 字段。 func (c *Config) timeout() time.Duration { if strings.TrimSpace(c.Timeout) == "" { return 60 * time.Second } d, err := time.ParseDuration(c.Timeout) if err != nil || d <= 0 { return 60 * time.Second } return d } // Engine 把配置、桶客户端与写入通道绑在一起。 type Engine struct { cfg Config bucket *BucketClient writer Writer } // New 构造 Engine。桶只读客户端必建;写入通道按 writer 配置选,选不出来也不 // 立刻失败——只读(pull/status)依然可用,真正需要写时 Apply 会报错。 func New(cfg Config) (*Engine, error) { cfg.Normalize() if cfg.Bucket == "" { return nil, fmt.Errorf("sync.bucket 未配置") } token, err := cfg.resolveToken() if err != nil { return nil, err } cfg.Token = token e := &Engine{ cfg: cfg, bucket: NewBucketClient(cfg.Endpoint, cfg.Bucket, token, cfg.timeout(), cfg.InsecureSkipVerify, cfg.CABundle), } if w, werr := NewWriter(cfg); werr == nil { e.writer = w } return e, nil } // Bucket 暴露只读客户端(cmd/sync 读远端 api_key 时用)。 func (e *Engine) Bucket() *BucketClient { return e.bucket } // WriterName 返回当前写入通道名(未选出时返回空串)。 func (e *Engine) WriterName() string { if e.writer == nil { return "" } return e.writer.Name() } // Config 返回归一化后的配置副本。 func (e *Engine) Config() Config { return e.cfg } // resolveToken 取 token:字段优先,其次 token_file,其次 HF_TOKEN 环境变量。 func (c *Config) resolveToken() (string, error) { if t := strings.TrimSpace(c.Token); t != "" { return t, nil } if fp := strings.TrimSpace(c.TokenFile); fp != "" { raw, err := readTrimmedFile(fp) if err != nil { return "", err } return raw, nil } return "", nil // 允许匿名读公开桶;写的时候才会因 401 报错 } // FetchRemoteAPIKey 从桶里的 config.json 读取中转网关的 api_key。 // // 存在的意义:relay 写入通道要带 Bearer,而那个 key 就躺在(私有的)桶里。 // 桶已经要凭 token 才能读了,再去手工抄一遍 key 没有意义。 func (e *Engine) FetchRemoteAPIKey(ctx context.Context) (string, error) { raw, err := e.bucket.Get(ctx, "config.json") if err != nil { return "", err } var cfg struct { APIKey string `json:"api_key"` } if err := json.Unmarshal(raw, &cfg); err != nil { return "", fmt.Errorf("解析远端 config.json 失败: %w", err) } if strings.TrimSpace(cfg.APIKey) == "" { return "", fmt.Errorf("远端 config.json 里 api_key 为空") } return cfg.APIKey, nil } // --------------------------------------------------------------------------- // 对账 // --------------------------------------------------------------------------- // Plan 计算这一轮要做什么,不产生任何副作用。 func (e *Engine) Plan(ctx context.Context, mode Mode) (*Plan, error) { local, err := e.scanLocal() if err != nil { return nil, fmt.Errorf("扫描本地失败: %w", err) } remote, err := e.collectRemote(ctx) if err != nil { return nil, err } p := buildPlan(mode, local, remote) p.Bucket = e.cfg.Bucket p.Writer = e.WriterName() return p, nil } // buildPlan 是纯函数形式的对账逻辑,便于单测覆盖各种组合。 func buildPlan(mode Mode, local, remote map[string]*FileState) *Plan { p := &Plan{Mode: mode, LocalFiles: len(local), RemoteFiles: len(remote)} paths := make([]string, 0, len(local)+len(remote)) seen := map[string]bool{} for k := range local { paths = append(paths, k) seen[k] = true } for k := range remote { if !seen[k] { paths = append(paths, k) } } sort.Strings(paths) for _, path := range paths { l, r := local[path], remote[path] p.Actions = append(p.Actions, decideAction(mode, path, l, r)) } return p } // decideAction 判定单个路径的动作。 func decideAction(mode Mode, path string, l, r *FileState) Action { category := catAuth if strings.HasPrefix(path, "data/") { category = catData } a := Action{Path: path, Category: category, Local: l, Remote: r} switch { case l == nil && r == nil: a.Op = OpNoop a.Reason = "两端均不存在(不应发生)" case r == nil: if mode.allowsUpload() { a.Op, a.Reason = OpUpload, "远端缺失,本端独有" } else { a.Op, a.Reason = OpNoop, "本端独有;当前模式不允许写远端" } case l == nil: if mode.allowsDownload() { a.Op, a.Reason = OpDownload, "本地缺失,远端独有" } else { a.Op, a.Reason = OpNoop, "远端独有;当前模式不允许写本地" } default: if l.SHA256 == r.SHA256 { a.Op, a.Reason = OpNoop, "两端内容一致" return a } a.Conflict = true var winner, reason string if category == catAuth { winner, reason = decideAuthWinner(l, r) } else { winner, reason = decideDataWinner(l, r) } switch { case winner == sideLocal && mode.allowsUpload(): a.Op, a.Reason = OpUpload, reason case winner == sideRemote && mode.allowsDownload(): a.Op, a.Reason = OpDownload, reason case winner == sideLocal: a.Op, a.Reason = OpNoop, reason+";但当前模式不允许写远端" default: a.Op, a.Reason = OpNoop, reason+";但当前模式不允许写本地" } } return a } // collectRemote 列出桶内候选文件并下载内容。 // // 为什么要下载才算得出结论:桶 tree 里的 xetHash 是 Xet 的内容哈希, // **不是**文件的 SHA-256(已实测:同一文件两者不同)。所以"是否一致"只能靠 // 拉下来自己算。候选集很小(凭证各 ~2.3KB、可选数据 ~2.7KB),这个代价可以忽略; // 反过来,用 xetHash 当 SHA-256 会误判成"永远不一致",每轮都白传一遍。 func (e *Engine) collectRemote(ctx context.Context) (map[string]*FileState, error) { files, err := e.bucket.List(ctx, "") if err != nil { return nil, fmt.Errorf("列出桶目录失败: %w", err) } out := map[string]*FileState{} for _, rf := range files { if !e.isCandidate(rf.Path) { continue } raw, err := e.bucket.Get(ctx, rf.Path) if err != nil { return nil, fmt.Errorf("下载 %s 失败: %w", rf.Path, err) } category := catAuth if strings.HasPrefix(rf.Path, "data/") { category = catData } out[rf.Path] = newFileState(rf.Path, raw, rf.Mtime, category) } return out, nil } // isCandidate 判定桶内一个路径是否属于同步范围。 // // 明确排除: // - config.json —— 两端的 listen / auth_dir / state_file / api_key 本来就不同, // 互相覆盖会把对端跑不起来。配置是"每端自己的",不是要同步的数据。 // - data/request-logs/** —— 本机请求归档,搬来搬去只撑大桶。 func (e *Engine) isCandidate(relPath string) bool { p := strings.ReplaceAll(relPath, `\`, "/") if p == "config.json" || strings.HasPrefix(p, "data/request-logs/") { return false } return PathAllowed(p, e.cfg.IncludeData) } // --------------------------------------------------------------------------- // 执行 // --------------------------------------------------------------------------- // Apply 执行计划。dryRun 为真时只回填"将会做什么",不落盘也不上传。 func (e *Engine) Apply(ctx context.Context, p *Plan, dryRun bool) (*Result, error) { res := &Result{Mode: p.Mode} // 批量上传:relay 通道一次 HTTP 调用推完,省掉冷启动后的多次往返。 var batch []PushFile batchable := false if !dryRun { if _, ok := e.writer.(interface { PutBatch(context.Context, []PushFile, bool) error }); ok { batchable = true } } for i := range p.Actions { a := &p.Actions[i] switch a.Op { case OpNoop: res.Skipped++ continue case OpDownload: if dryRun { a.Applied = true res.Downloaded++ continue } if err := e.applyDownload(ctx, a, res); err != nil { a.Err = err.Error() res.Errors = append(res.Errors, fmt.Sprintf("%s: %v", a.Path, err)) res.Failed++ continue } a.Applied = true res.Downloaded++ case OpUpload: if e.writer == nil { err := fmt.Errorf("没有可用的写入通道(配置 sync.writer / relay_url,或安装 hf CLI)") a.Err = err.Error() res.Errors = append(res.Errors, fmt.Sprintf("%s: %v", a.Path, err)) res.Failed++ continue } if dryRun { a.Applied = true res.Uploaded++ continue } if batchable { batch = append(batch, PushFile{Path: a.Path, Content: a.Local.Content, Mtime: a.Local.Mtime}) a.Applied = true res.Uploaded++ continue } if err := e.writer.Put(ctx, a.Path, a.Local.Content); err != nil { a.Err = err.Error() res.Errors = append(res.Errors, fmt.Sprintf("%s: %v", a.Path, err)) res.Failed++ continue } a.Applied = true res.Uploaded++ } } if len(batch) > 0 { type batchWriter interface { PutBatch(context.Context, []PushFile, bool) error } if bw, ok := e.writer.(batchWriter); ok { if err := bw.PutBatch(ctx, batch, true); err != nil { // 批量失败:把已计入的动作全部回落为失败,避免"报告成功其实没上去"。 res.Errors = append(res.Errors, err.Error()) failed := map[string]bool{} for _, bf := range batch { failed[bf.Path] = true } for i := range p.Actions { if p.Actions[i].Op == OpUpload && failed[p.Actions[i].Path] { p.Actions[i].Applied = false p.Actions[i].Err = err.Error() res.Uploaded-- res.Failed++ } } } } } res.Actions = p.Actions return res, nil } // applyDownload 把远端内容落到本地。 func (e *Engine) applyDownload(ctx context.Context, a *Action, res *Result) error { content := a.Remote.Content if content == nil { raw, err := e.bucket.Get(ctx, a.Path) if err != nil { return err } content = raw } fp, err := e.localPath(a.Path) if err != nil { return err } if err := writeLocalAtomic(fp, content, a.Remote.Mtime); err != nil { return err } _ = res return nil } // Sync 是 Plan + Apply 的便捷组合。 func (e *Engine) Sync(ctx context.Context, mode Mode, dryRun bool) (*Plan, *Result, error) { p, err := e.Plan(ctx, mode) if err != nil { return nil, nil, err } res, err := e.Apply(ctx, p, dryRun) if err != nil { return p, res, err } return p, res, nil }