Coverage for app/services/build_lock.py: 98.61%

1"""Redis-backed distributed lock around the Packer image build. 

2 

3Why this exists: the build phase in `tasks.py` does a check-then-act pair 

4(`check_image_exists` → `packer.build`) on the shared OpenStack Glance store. 

5Two parallel workers triggering a build of the same `(project_id, image_name)` 

6both observe "not found" and both kick off a build, leaving a duplicate image 

7behind and burning ~10 minutes of compute. This lock serializes the build for 

8a given image name within a given OpenStack project so only one worker 

9actually builds; the other re-checks Glance after waiting and skips straight 

10to Terraform if the image now exists. 

11 

12Backend: Redis (already present as Celery's result backend, no new dep). 

13 

14Lock semantics: 

15 

16* ``SET NX PX`` with a unique token holds the lock with a short TTL 

17 (default 5 min). A daemon thread inside the holder periodically 

18 refreshes the TTL via ``PEXPIRE`` so a long-running build can't 

19 outlive its lease. 

20* Release uses a Lua-CAS so a worker can never release a lock that 

21 another worker grabbed after the first one's TTL expired. 

22* If the holding worker process is killed before ``release()`` runs 

23 (SIGKILL, OOM, container restart), the heartbeat thread dies with 

24 it and the lock self-heals after ``ttl_ms`` — bounded staleness. 

25""" 

26 

27from __future__ import annotations 

28 

29import threading 

30import time 

31import uuid 

32 

33import redis 

34 

35from ..config import settings 

36from ..utils.logger import get_logger 

37 

38logger = get_logger(__name__) 

39 

40 

41_LUA_RELEASE = """ 

42if redis.call("get", KEYS[1]) == ARGV[1] then 

43 return redis.call("del", KEYS[1]) 

44end 

45return 0 

46""" 

47 

48_LUA_RENEW = """ 

49if redis.call("get", KEYS[1]) == ARGV[1] then 

50 return redis.call("pexpire", KEYS[1], ARGV[2]) 

51end 

52return 0 

53""" 

54 

55# Lock lease. Short enough that a crashed worker doesn't block the next 

56# deploy for long; the heartbeat below extends it on every tick so an 

57# in-flight build that takes longer than the lease still keeps the lock. 

58_DEFAULT_TTL_MS = 5 * 60 * 1000 

59# Heartbeat interval. Must be < TTL with margin so we don't miss a 

60# refresh window after a momentary GC pause / Redis hiccup. 

61_DEFAULT_HEARTBEAT_S = 60 

62_DEFAULT_POLL_S = 5 

63_DEFAULT_TOTAL_WAIT_S = 25 * 60 

64 

65 

66def _redis_client() -> redis.Redis: 

67 return redis.Redis.from_url(settings.CELERY_RESULT_BACKEND) 

68 

69 

70class PackerBuildLock: 

71 """Acquire a Redis lock keyed on (project, image_name). 

72 

73 The class is a context-manager-shaped polling loop. Pattern: 

74 

75 lock = PackerBuildLock(project_id, image_name) 

76 try: 

77 while True: 

78 held = lock.acquire_or_wait() 

79 if held: 

80 if image_already_exists(): break 

81 packer.build(...) 

82 break 

83 if image_already_exists(): break 

84 finally: 

85 lock.release() 

86 """ 

87 

88 def __init__( 

89 self, 

90 project_id: str, 

91 image_name: str, 

92 *, 

93 ttl_ms: int = _DEFAULT_TTL_MS, 

94 heartbeat_interval_s: int = _DEFAULT_HEARTBEAT_S, 

95 poll_interval_s: int = _DEFAULT_POLL_S, 

96 total_wait_s: int = _DEFAULT_TOTAL_WAIT_S, 

97 ): 

98 self.key = f"lock:packer:{project_id or 'unknown'}:{image_name}" 

99 self.token = uuid.uuid4().hex 

100 self.ttl_ms = ttl_ms 

101 self.heartbeat_interval_s = heartbeat_interval_s 

102 self.poll_interval_s = poll_interval_s 

103 self.deadline = time.monotonic() + total_wait_s 

104 self._client = _redis_client() 

105 self._held = False 

106 # Heartbeat coordination. A daemon thread runs while the lock is 

107 # held and renews the TTL on every interval; ``_stop`` flips it 

108 # off during ``release()``. Daemon=True so the thread doesn't 

109 # keep the worker alive on shutdown. 

110 self._heartbeat: threading.Thread | None = None 

111 self._stop = threading.Event() 

112 

113 # ----- acquisition --------------------------------------------------- 

114 

115 def acquire_or_wait(self) -> bool: 

116 """Try to acquire. Returns True if held, False if we slept and the 

117 caller should re-check Glance and call again. Raises TimeoutError 

118 if the total wait budget is exhausted.""" 

119 if self._client.set(self.key, self.token, nx=True, px=self.ttl_ms): 

120 self._held = True 

121 self._start_heartbeat() 

122 logger.info("Acquired Packer build lock", lock_key=self.key, ttl_ms=self.ttl_ms) 

123 return True 

124 if time.monotonic() > self.deadline: 

125 raise TimeoutError( 

126 f"Timed out waiting for Packer build lock {self.key} " f"(another worker is still building this image)" 

127 ) 

128 # Look up the lock's remaining TTL so the operator/log reader has 

129 # a hint at how long the wait will be. ``pttl`` returns -2 if the 

130 # key has just expired (race with another waiter), -1 if no TTL 

131 # is set (shouldn't happen — we always set PX), or the remaining 

132 # ms otherwise. 

133 ttl_left_ms = self._client.pttl(self.key) 

134 logger.info( 

135 "Waiting for in-progress Packer build", 

136 lock_key=self.key, 

137 poll_interval_s=self.poll_interval_s, 

138 other_holder_ttl_ms=ttl_left_ms if ttl_left_ms and ttl_left_ms > 0 else None, 

139 ) 

140 time.sleep(self.poll_interval_s) 

141 return False 

142 

143 # ----- heartbeat ----------------------------------------------------- 

144 

145 def _start_heartbeat(self) -> None: 

146 """Spawn the renewal thread. 

147 

148 The thread loops on ``_stop.wait(interval)`` so it sleeps 

149 cancellably and exits promptly when ``release()`` runs. CAS via 

150 the Lua script ensures we only renew our own lock — if another 

151 worker has somehow taken over after our TTL expired (e.g. our 

152 heartbeat fell behind a Redis outage), we don't accidentally 

153 steal their lease back. 

154 """ 

155 self._stop.clear() 

156 thread = threading.Thread( 

157 target=self._heartbeat_loop, 

158 name=f"packer-lock-heartbeat-{self.key[-12:]}", 

159 daemon=True, 

160 ) 

161 thread.start() 

162 self._heartbeat = thread 

163 

164 def _heartbeat_loop(self) -> None: 

165 while not self._stop.wait(self.heartbeat_interval_s): 

166 try: 

167 renewed = self._client.eval(_LUA_RENEW, 1, self.key, self.token, self.ttl_ms) 

168 except Exception as e: 

169 # Redis blip — try again next tick. We don't break the 

170 # loop on a single failure; if Redis stays down past 

171 # our TTL the lock will expire and another worker will 

172 # eventually pick up. 

173 logger.warning(f"Packer lock renewal raised: {e}") 

174 continue 

175 if not renewed: 

176 # Lock is gone (TTL expired faster than we could renew, 

177 # or someone manually deleted it). Stop heartbeating — 

178 # ``release()`` is a no-op in this case. 

179 logger.warning( 

180 "Packer lock vanished mid-build; stopping heartbeat", 

181 lock_key=self.key, 

182 ) 

183 self._held = False 

184 return 

185 

186 # ----- release ------------------------------------------------------- 

187 

188 def release(self) -> None: 

189 if not self._held: 

190 return 

191 # Stop the heartbeat first so it doesn't race with the delete. 

192 self._stop.set() 

193 if self._heartbeat is not None: 

194 self._heartbeat.join(timeout=2) 

195 self._heartbeat = None 

196 try: 

197 self._client.eval(_LUA_RELEASE, 1, self.key, self.token) 

198 except Exception as e: 

199 # Best-effort: even if release fails, the TTL will eventually 

200 # expire and the lock self-heals. 

201 logger.warning(f"Failed to release Packer lock {self.key}: {e}") 

202 finally: 

203 self._held = False 

204 

205 # ----- context manager ergonomics ----------------------------------- 

206 

207 def __enter__(self) -> PackerBuildLock: 

208 return self 

209 

210 def __exit__(self, *exc_info: object) -> None: 

211 self.release()