From ee22cb8ad0ca13ed3a44f1f22b173501cb01642b Mon Sep 17 00:00:00 2001 From: tamakyi Date: Tue, 8 Sep 2026 19:13:24 +0800 Subject: [PATCH] =?UTF-8?q?git-sync-web=20=E8=87=AA=E5=8A=A8=E6=8F=90?= =?UTF-8?q?=E4=BA=A4=202026-09-08=2019:13:24?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- engine.go | 183 +++++++++++++++++++++++++++++++---------- logs/git-auto-sync.log | 3 + server.go | 19 +++++ 3 files changed, 160 insertions(+), 45 deletions(-) diff --git a/engine.go b/engine.go index a119f20..0169aa8 100644 --- a/engine.go +++ b/engine.go @@ -16,8 +16,7 @@ import ( "time" ) -// ---------- 事件总线（SSE 广播） ---------- - +// ---------- 事件总线 ---------- type EventBus struct { mu sync.Mutex next int @@ -50,16 +49,15 @@ func (b *EventBus) Publish(data []byte) { for _, ch := range b.subs { select { case ch <- data: - default: // 订阅者阻塞时丢弃，避免拖慢同步循环 + default: } } } // ---------- 状态 ---------- - type SyncResult struct { Time string `json:"time"` - Result string `json:"result"` // ok | error | conflict + Result string `json:"result"` // ok | error | conflict | push_conflict Detail string `json:"detail"` } @@ -78,13 +76,17 @@ type RepoSnapshot struct { Cron string `json:"cron"` Remotes []Remote `json:"remotes"` Running bool `json:"running"` - Status string `json:"status"` + Status string `json:"status"` // idle | ok | error | conflict | push_conflict Ahead int `json:"ahead"` Behind int `json:"behind"` Dirty int `json:"dirty"` Conflicts []string `json:"conflicts"` LastError string `json:"last_error"` LastSync *SyncResult `json:"last_sync"` + // 新增：推送冲突信息 + PushConflict bool `json:"push_conflict"` + PushConflictRemote string `json:"push_conflict_remote"` + PushConflictBranch string `json:"push_conflict_branch"` } type repoState struct { @@ -92,7 +94,7 @@ type repoState struct { cfg *RepoConfig running bool syncing bool - status string // idle | ok | error | conflict + status string // idle | ok | error | conflict | push_conflict ahead int behind int dirty int @@ -100,6 +102,10 @@ type repoState struct { lastError string lastSync *SyncResult stopCh chan struct{} + // 推送冲突信息 + pushConflict bool + pushConflictRemote string + pushConflictBranch string } type Engine struct { @@ -107,10 +113,9 @@ type Engine struct { cfgDir string logDir string statsDir string - - mu sync.Mutex - states map[string]*repoState - bus *EventBus + mu sync.Mutex + states map[string]*repoState + bus *EventBus } func NewEngine(base string) (*Engine, error) { @@ -164,7 +169,6 @@ func (e *Engine) sortedNames() []string { return names } -// FIX: LoadAll 加锁，避免并发读写 map func (e *Engine) LoadAll() { e.mu.Lock() defer e.mu.Unlock() @@ -251,6 +255,9 @@ func (s *repoState) snapshot(live bool) RepoSnapshot { Conflicts: append([]string(nil), s.conflicts...), LastError: s.lastError, LastSync: s.lastSync, + PushConflict: s.pushConflict, + PushConflictRemote: s.pushConflictRemote, + PushConflictBranch: s.pushConflictBranch, } s.mu.Unlock() if live { @@ -286,7 +293,6 @@ func (e *Engine) pushStatus(s *repoState) { } // ---------- 启动 / 停止 ---------- - func (e *Engine) Start(name string) error { s := e.get(name) if s == nil { @@ -401,7 +407,6 @@ func (e *Engine) SyncNow(name string) error { } // ---------- 同步核心 ---------- - func (e *Engine) withRetry(s *repoState, label string, fn func() (string, error)) (string, error) { maxAttempts, sleep, retryCount := 1, 0, 0 s.mu.Lock() @@ -448,7 +453,7 @@ func (e *Engine) setOK(s *repoState, detail string) { e.pushStatus(s) } -// ensureRemotes 把配置中的远程同步到仓库（缺失则添加，URL 变化则更新） +// ensureRemotes 把配置中的远程同步到仓库 func (e *Engine) ensureRemotes(s *repoState, cfg *RepoConfig) { for _, r := range cfg.Remotes { out, err := gitRun(cfg.RepoPath, 15*time.Second, "remote", "get-url", r.Name) @@ -480,7 +485,7 @@ func (e *Engine) refreshCounts(s *repoState, cfg *RepoConfig, remote string) { e.pushStatus(s) } -// safeSync 包装 syncOnce：捕获后台 goroutine 的 panic，避免整个进程被杀死。 +// safeSync 包装 syncOnce：捕获 panic func (e *Engine) safeSync(s *repoState, trigger string) { defer func() { if r := recover(); r != nil { @@ -495,7 +500,7 @@ func (e *Engine) safeSync(s *repoState, trigger string) { e.syncOnce(s, trigger) } -// FIX: 调整同步顺序：先拉取后推送；分支切换避免强制覆盖；pull 使用重试 +// syncOnce 核心同步逻辑（已调整顺序，并增加 push_conflict 处理） func (e *Engine) syncOnce(s *repoState, trigger string) { s.mu.Lock() if s.syncing { @@ -531,19 +536,17 @@ func (e *Engine) syncOnce(s *repoState, trigger string) { return } - // FIX: 分支对齐逻辑 – 避免使用 -B 强制覆盖，本地存在则普通切换，不存在则创建 + // 分支对齐（安全切换） if cur, cerr := currentBranch(cfg.RepoPath); cerr == nil && cur != cfg.Branch { tip, remoteHasBranch := remoteTip(cfg.RepoPath, remote, cfg.Branch) localExists := branchExists(cfg.RepoPath, cfg.Branch) if localExists { - // 本地分支存在，普通切换（可能因未提交改动而失败） if _, err := gitRun(cfg.RepoPath, 30*time.Second, "checkout", cfg.Branch); err != nil { e.setFail(s, fmt.Sprintf("切换到配置分支 %s 失败，请先提交或暂存改动: %v", cfg.Branch, err)) return } e.logf(s, "info", "本地分支与配置不一致（%s → %s），已自动切换", cur, cfg.Branch) } else { - // 本地分支不存在，创建（不会覆盖任何内容） if remoteHasBranch { if _, err := gitRun(cfg.RepoPath, 30*time.Second, "checkout", "-b", cfg.Branch, tip); err != nil { e.setFail(s, fmt.Sprintf("基于远程 %s/%s 创建分支失败: %v", remote, cfg.Branch, err)) @@ -561,7 +564,7 @@ func (e *Engine) syncOnce(s *repoState, trigger string) { } } - // 自动提交（若有未提交改动） + // 自动提交 dirty := dirtyCount(cfg.RepoPath) if dirty > 0 { if cfg.AutoCommit { @@ -582,7 +585,7 @@ func (e *Engine) syncOnce(s *repoState, trigger string) { } } - // 计算初始 ahead/behind + // 计算 ahead/behind var ahead, behind int var err error tip, remoteHasBranch := remoteTip(cfg.RepoPath, remote, cfg.Branch) @@ -593,7 +596,6 @@ func (e *Engine) syncOnce(s *repoState, trigger string) { return } } else { - // 远程分支不存在：认为所有本地提交领先 if out, cerr := gitRun(cfg.RepoPath, 15*time.Second, "rev-list", "--count", "HEAD"); cerr == nil { ahead, _ = strconv.Atoi(strings.TrimSpace(out)) } @@ -604,12 +606,11 @@ func (e *Engine) syncOnce(s *repoState, trigger string) { s.mu.Unlock() e.pushStatus(s) - // FIX: 先拉取（pull），再推送（push） + // ---- 先拉取 ---- if behind > 0 && (cfg.Mode == "pull" || cfg.Mode == "both") { if !e.pullRemote(s, cfg, remote, behind) { return } - // 拉取后重新计算 ahead/behind ahead, behind, _ = aheadBehind(cfg.RepoPath, remote, cfg.Branch) s.mu.Lock() s.ahead, s.behind = ahead, behind @@ -617,14 +618,25 @@ func (e *Engine) syncOnce(s *repoState, trigger string) { e.pushStatus(s) } + // ---- 再推送 ---- if ahead > 0 && (cfg.Mode == "push" || cfg.Mode == "both") { - if !e.pushAll(s, cfg, ahead) { - e.setFail(s, "推送失败") + // pushAll 可能返回错误，如果是 non-fast-forward，设置 push_conflict 状态 + pushOK, pushErr := e.pushAll(s, cfg, ahead) + if !pushOK { + // 检查是否是 non-fast-forward 错误 + if pushErr != nil && strings.Contains(strings.ToLower(pushErr.Error()), "non-fast-forward") { + // 记录冲突远程（取第一个冲突的远程） + // 在 pushAll 内部已经设置了 pushConflict 状态，这里只需退出 + e.logf(s, "warn", "推送被拒绝（non-fast-forward），等待用户决策") + return + } + // 其他错误 + e.setFail(s, "推送失败: "+pushErr.Error()) return } } - // 最终刷新状态 + // 最终刷新 ahead, behind, _ = aheadBehind(cfg.RepoPath, remote, cfg.Branch) now := time.Now().Format("2006-01-02 15:04:05") s.mu.Lock() @@ -632,13 +644,20 @@ func (e *Engine) syncOnce(s *repoState, trigger string) { s.status = "ok" s.lastError = "" s.lastSync = &SyncResult{Time: now, Result: "ok", Detail: "同步完成"} + // 清除可能遗留的推送冲突状态 + s.pushConflict = false + s.pushConflictRemote = "" + s.pushConflictBranch = "" s.mu.Unlock() e.logf(s, "info", "=== 同步完成（本地领先 %d，落后 %d）===", ahead, behind) e.pushStatus(s) } -func (e *Engine) pushAll(s *repoState, cfg *RepoConfig, n int) bool { +// pushAll 返回 (是否全部成功, 错误信息) +// 如果遇到 non-fast-forward，会设置 pushConflict 状态并返回 false，错误包含 "non-fast-forward" +func (e *Engine) pushAll(s *repoState, cfg *RepoConfig, n int) (bool, error) { allOK := true + var firstErr error for _, r := range cfg.Remotes { _, err := e.withRetry(s, "push → "+r.Name, func() (string, error) { return gitRun(cfg.RepoPath, 300*time.Second, "push", r.Name, cfg.Branch) @@ -646,19 +665,38 @@ func (e *Engine) pushAll(s *repoState, cfg *RepoConfig, n int) bool { if err != nil { e.logf(s, "error", "推送至 %s 失败: %v", r.Name, err) allOK = false + if firstErr == nil { + firstErr = err + } + // 检测到 non-fast-forward，记录状态并继续尝试其他远程 + if strings.Contains(strings.ToLower(err.Error()), "non-fast-forward") { + s.mu.Lock() + s.status = "push_conflict" + s.pushConflict = true + s.pushConflictRemote = r.Name + s.pushConflictBranch = cfg.Branch + s.lastError = "推送被拒绝（non-fast-forward），请选择强制覆盖或放弃" + s.lastSync = &SyncResult{ + Time: time.Now().Format("2006-01-02 15:04:05"), + Result: "push_conflict", + Detail: fmt.Sprintf("远程 %s 有新提交，推送被拒绝", r.Name), + } + s.mu.Unlock() + e.pushStatus(s) + } } else { e.logf(s, "info", "推送至 %s 成功（%d 个提交）", r.Name, n) } } if allOK { e.stat(s, "push", "success", n) - } else { - e.stat(s, "push", "failed", n) + return true, nil } - return allOK + e.stat(s, "push", "failed", n) + return false, firstErr } -// FIX: pullRemote 使用 withRetry 包裹主 pull 命令 +// pullRemote 已使用重试 func (e *Engine) pullRemote(s *repoState, cfg *RepoConfig, remote string, n int) bool { args := []string{"pull"} if cfg.ConflictStrategy == "rebase" { @@ -673,7 +711,6 @@ func (e *Engine) pullRemote(s *repoState, cfg *RepoConfig, remote string, n int) e.logf(s, "info", "拉取成功（%d 个提交）", n) return true } - // 处理冲突 files, cerr := conflictFiles(cfg.RepoPath) if cerr == nil && len(files) > 0 { if cfg.ConflictStrategy == "abort" { @@ -699,8 +736,65 @@ func (e *Engine) pullRemote(s *repoState, cfg *RepoConfig, remote string, n int) return false } -// ---------- 手动操作 ---------- +// ---------- 推送冲突处理 ---------- +// ForcePush 强制推送（--force-with-lease）到冲突的远程 +func (e *Engine) ForcePush(name string) error { + s := e.get(name) + if s == nil { + return fmt.Errorf("配置不存在: %s", name) + } + s.mu.Lock() + if !s.pushConflict { + s.mu.Unlock() + return fmt.Errorf("当前没有待处理的推送冲突") + } + remote := s.pushConflictRemote + branch := s.pushConflictBranch + path := s.cfg.RepoPath + s.mu.Unlock() + e.logf(s, "info", "执行强制推送至 %s %s", remote, branch) + _, err := gitRun(path, 300*time.Second, "push", "--force-with-lease", remote, branch) + if err != nil { + e.logf(s, "error", "强制推送失败: %v", err) + return err + } + e.logf(s, "info", "强制推送成功") + s.mu.Lock() + s.pushConflict = false + s.pushConflictRemote = "" + s.pushConflictBranch = "" + s.status = "ok" + s.lastError = "" + s.lastSync = &SyncResult{Time: time.Now().Format("2006-01-02 15:04:05"), Result: "ok", Detail: "强制推送完成"} + s.mu.Unlock() + e.pushStatus(s) + return nil +} + +// AbortPush 放弃强制推送，清除冲突状态 +func (e *Engine) AbortPush(name string) error { + s := e.get(name) + if s == nil { + return fmt.Errorf("配置不存在: %s", name) + } + s.mu.Lock() + defer s.mu.Unlock() + if !s.pushConflict { + return fmt.Errorf("当前没有待处理的推送冲突") + } + s.pushConflict = false + s.pushConflictRemote = "" + s.pushConflictBranch = "" + s.status = "ok" + s.lastError = "用户放弃强制推送" + s.lastSync = &SyncResult{Time: time.Now().Format("2006-01-02 15:04:05"), Result: "ok", Detail: "放弃强制推送，状态已清除"} + e.logf(s, "warn", "放弃强制推送") + e.pushStatus(s) + return nil +} + +// ---------- 手动操作 ---------- func (e *Engine) ManualGit(name, action string) error { s := e.get(name) if s == nil { @@ -739,7 +833,6 @@ func (e *Engine) ManualGit(name, action string) error { e.logf(s, "info", "fetch 完成") e.refreshCounts(s, cfg, remote) return nil - case "push": e.logf(s, "info", "手动 push…") ahead, behind, _ := aheadBehind(cfg.RepoPath, remote, cfg.Branch) @@ -751,14 +844,14 @@ func (e *Engine) ManualGit(name, action string) error { e.refreshCounts(s, cfg, remote) return nil } - if !e.pushAll(s, cfg, ahead) { - e.setFail(s, "推送失败") - return fmt.Errorf("推送失败") + ok, err := e.pushAll(s, cfg, ahead) + if !ok { + e.logf(s, "error", "推送失败: %v", err) + return err } e.refreshCounts(s, cfg, remote) e.setOK(s, "手动推送完成") return nil - case "pull": e.logf(s, "info", "手动 pull…") if _, err := gitRun(cfg.RepoPath, 90*time.Second, "fetch", remote); err != nil { @@ -782,7 +875,6 @@ func (e *Engine) ManualGit(name, action string) error { } // ---------- 冲突处理 ---------- - func (e *Engine) ConflictAbort(name string) error { s := e.get(name) if s == nil { @@ -836,7 +928,6 @@ func (e *Engine) ConflictResolve(name string) error { } // ---------- 配置 CRUD ---------- - func (e *Engine) Create(c *RepoConfig) error { if err := validateRepoConfig(e.cfgDir, c, true); err != nil { return err @@ -877,6 +968,10 @@ func (e *Engine) Update(c *RepoConfig) error { s.lastError = "" s.ahead, s.behind, s.dirty = 0, 0, 0 s.lastSync = nil + // 清除推送冲突 + s.pushConflict = false + s.pushConflictRemote = "" + s.pushConflictBranch = "" s.mu.Unlock() e.logf(s, "info", "配置已更新") if wasRunning { @@ -903,8 +998,7 @@ func (e *Engine) Delete(name string) error { return nil } -// ---------- 日志 / 统计 / 冲突 diff 读取 ---------- - +// ---------- 日志 / 统计 / 冲突 diff ---------- type StatRow struct { Time string `json:"time"` Action string `json:"action"` @@ -985,7 +1079,6 @@ func (e *Engine) ConflictDiffs(name string) []ConflictFile { } // ---------- 提交历史 ---------- - var hashRe = regexp.MustCompile(`^[0-9a-fA-F]{6,40}$`) type Commit struct { diff --git a/logs/git-auto-sync.log b/logs/git-auto-sync.log index 1065686..8d407d4 100644 --- a/logs/git-auto-sync.log +++ b/logs/git-auto-sync.log @@ -328,3 +328,6 @@ hint: its remote counterpart. If you want to integrate the remote changes, hint: use 'git pull' before pushing again. hint: See the 'Note about fast-forwards' in 'git push --help' for details. [2026-09-08 18:55:22] 推送失败，本轮同步已终止（跳过拉取以避免分叉加剧） +[2026-09-08 19:13:17] 监控已启动 +[2026-09-08 19:13:17] === 开始同步（启动检查，模式 push）=== +[2026-09-08 19:13:24] 检测到 3 个未提交改动，执行自动提交 diff --git a/server.go b/server.go index 93fb34f..26b7e4b 100644 --- a/server.go +++ b/server.go @@ -43,6 +43,8 @@ func NewServer(e *Engine, static fs.FS, password string) *Server { mux.HandleFunc("POST /api/repos/{name}/stop", s.wrapRepo(s.stopRepo)) mux.HandleFunc("POST /api/repos/{name}/sync", s.wrapRepo(s.syncRepo)) mux.HandleFunc("POST /api/repos/{name}/action", s.wrapRepo(s.actionRepo)) + mux.HandleFunc("POST /api/repos/{name}/force-push", s.wrapRepo(s.forcePushRepo)) + mux.HandleFunc("POST /api/repos/{name}/abort-push", s.wrapRepo(s.abortPushRepo)) mux.HandleFunc("GET /api/repos/{name}/conflicts", s.wrapRepo(s.conflictsRepo)) mux.HandleFunc("GET /api/repos/{name}/log", s.wrapRepo(s.logRepo)) mux.HandleFunc("GET /api/repos/{name}/stats", s.wrapRepo(s.statsRepo)) @@ -52,6 +54,7 @@ func NewServer(e *Engine, static fs.FS, password string) *Server { return s } + func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { // 访问密码保护：静态资源开放，/api/* 需要会话（登录接口除外） if s.password != "" && strings.HasPrefix(r.URL.Path, "/api/") && @@ -311,3 +314,19 @@ func (s *Server) handleEvents(w http.ResponseWriter, r *http.Request) { } } } + +func (s *Server) forcePushRepo(w http.ResponseWriter, r *http.Request, name string) { + if err := s.engine.ForcePush(name); err != nil { + writeErr(w, 400, err.Error()) + return + } + writeJSON(w, 200, map[string]any{"ok": true}) +} + +func (s *Server) abortPushRepo(w http.ResponseWriter, r *http.Request, name string) { + if err := s.engine.AbortPush(name); err != nil { + writeErr(w, 400, err.Error()) + return + } + writeJSON(w, 200, map[string]any{"ok": true}) +} \ No newline at end of file