Coverage for little_loops / parallel / git_lock.py: 25%

1"""Thread-safe git operations with retry logic for parallel processing. 

2 

3Provides a shared lock for git operations on the main repository to prevent 

4concurrent operations from conflicting over .git/index.lock. 

5 

6The index.lock race condition occurs when: 

7- MergeCoordinator runs stash/merge/pull operations 

8- WorkerPool captures baseline git status or cleans up leaked files 

9- Both touch the main repo's git index simultaneously 

10 

11This module provides: 

12- A threading lock to serialize git operations on the main repo 

13- Retry logic with exponential backoff for transient index.lock errors 

14""" 

15 

16from __future__ import annotations 

17 

18import subprocess 

19import threading 

20import time 

21from pathlib import Path 

22from typing import TYPE_CHECKING 

23 

24if TYPE_CHECKING: 

25 from little_loops.logger import Logger 

26 

27 

28class GitLock: 

29 """Thread-safe wrapper for git operations on a repository. 

30 

31 Serializes all git operations through a single lock to prevent 

32 index.lock conflicts. Provides automatic retry with exponential 

33 backoff for transient lock errors. 

34 

35 Example: 

36 >>> git_lock = GitLock(logger) 

37 >>> # Use context manager for custom operations 

38 >>> with git_lock: 

39 ... subprocess.run(["git", "status"], cwd=repo_path) 

40 >>> # Or use the run method for automatic retry 

41 >>> result = git_lock.run(["status"], cwd=repo_path) 

42 """ 

43 

44 def __init__( 

45 self, 

46 logger: Logger | None = None, 

47 max_retries: int = 3, 

48 initial_backoff: float = 0.5, 

49 max_backoff: float = 8.0, 

50 ) -> None: 

51 """Initialize the git lock. 

52 

53 Args: 

54 logger: Optional logger for retry messages 

55 max_retries: Maximum number of retries on index.lock error 

56 initial_backoff: Initial backoff delay in seconds 

57 max_backoff: Maximum backoff delay in seconds 

58 """ 

59 # RLock allows same thread to acquire lock multiple times 

60 # (needed for nested git operations within a single method) 

61 self._lock = threading.RLock() 

62 self._logger = logger 

63 self.max_retries = max_retries 

64 self.initial_backoff = initial_backoff 

65 self.max_backoff = max_backoff 

66 

67 def __enter__(self) -> GitLock: 

68 """Acquire the lock for manual git operations.""" 

69 self._lock.acquire() 

70 return self 

71 

72 def __exit__( 

73 self, 

74 exc_type: type[BaseException] | None, 

75 exc_val: BaseException | None, 

76 exc_tb: object, 

77 ) -> None: 

78 """Release the lock.""" 

79 self._lock.release() 

80 

81 def run( 

82 self, 

83 args: list[str], 

84 cwd: Path, 

85 timeout: float = 30, 

86 capture_output: bool = True, 

87 text: bool = True, 

88 ) -> subprocess.CompletedProcess[str]: 

89 """Run a git command with lock and retry logic. 

90 

91 Args: 

92 args: Git command arguments (without 'git' prefix) 

93 cwd: Working directory for the command 

94 timeout: Command timeout in seconds 

95 capture_output: Whether to capture stdout/stderr 

96 text: Whether to decode output as text 

97 

98 Returns: 

99 CompletedProcess with command result 

100 """ 

101 with self._lock: 

102 return self._run_with_retry( 

103 args=args, 

104 cwd=cwd, 

105 timeout=timeout, 

106 capture_output=capture_output, 

107 text=text, 

108 ) 

109 

110 def _run_with_retry( 

111 self, 

112 args: list[str], 

113 cwd: Path, 

114 timeout: float, 

115 capture_output: bool, 

116 text: bool, 

117 ) -> subprocess.CompletedProcess[str]: 

118 """Run git command with retry on index.lock errors. 

119 

120 Args: 

121 args: Git command arguments (without 'git' prefix) 

122 cwd: Working directory 

123 timeout: Command timeout 

124 capture_output: Whether to capture output 

125 text: Whether to decode as text 

126 

127 Returns: 

128 CompletedProcess with result 

129 """ 

130 cmd = ["git"] + args 

131 backoff = self.initial_backoff 

132 last_result: subprocess.CompletedProcess[str] | None = None 

133 

134 for attempt in range(self.max_retries + 1): 

135 try: 

136 result = subprocess.run( 

137 cmd, 

138 cwd=cwd, 

139 timeout=timeout, 

140 capture_output=capture_output, 

141 text=text, 

142 ) 

143 

144 # Check for index.lock error 

145 if result.returncode != 0 and self._is_index_lock_error(result.stderr or ""): 

146 last_result = result 

147 if attempt < self.max_retries: 

148 if self._logger: 

149 self._logger.debug( 

150 f"Git index.lock conflict, retrying in {backoff:.1f}s " 

151 f"(attempt {attempt + 1}/{self.max_retries})" 

152 ) 

153 time.sleep(backoff) 

154 backoff = min(backoff * 2, self.max_backoff) 

155 continue 

156 

157 return result 

158 

159 except subprocess.TimeoutExpired: 

160 if attempt < self.max_retries: 

161 if self._logger: 

162 self._logger.debug( 

163 f"Git command timed out, retrying in {backoff:.1f}s " 

164 f"(attempt {attempt + 1}/{self.max_retries})" 

165 ) 

166 time.sleep(backoff) 

167 backoff = min(backoff * 2, self.max_backoff) 

168 else: 

169 raise 

170 

171 # Return last result if we exhausted retries 

172 if last_result is not None: 

173 if self._logger: 

174 self._logger.warning( 

175 f"Git command failed after {self.max_retries} retries: " 

176 f"{last_result.stderr[:200] if last_result.stderr else 'unknown error'}" 

177 ) 

178 return last_result 

179 

180 # Should not reach here, but provide fallback 

181 raise RuntimeError(f"Git command failed after {self.max_retries} retries") 

182 

183 @staticmethod 

184 def _is_index_lock_error(stderr: str) -> bool: 

185 """Check if error is due to index.lock conflict. 

186 

187 Args: 

188 stderr: Error output from git command 

189 

190 Returns: 

191 True if this is an index.lock error that may be retried 

192 """ 

193 if not stderr: 

194 return False 

195 

196 indicators = [ 

197 "index.lock", 

198 "Unable to create", 

199 "Another git process seems to be running", 

200 "File exists", 

201 ] 

202 

203 return any(indicator in stderr for indicator in indicators)