Coverage for dispatch/scripts.py: 100%

1"""Lua scripts — one per job state transition (spec §8, §9). 

2 

3Each script is atomic on the Redis side, so a state transition can never be 

4observed half-applied even under concurrent runners (INV2/INV3/INV4/INV5). 

5Every script takes ``now`` via ARGV — never ``redis.call('TIME')`` — so tests 

6are deterministic and fakeredis (which has no server clock for scripts) works. 

7 

8The scripts here cover slice 2 of the delivery plan (§15.2): ``claim_pending``, 

9``renew_lease``, ``reclaim_expired``. ``abort_requeue`` belongs to slice 3 and 

10is deliberately absent. 

11 

12Key handling: static keys (``jobs:pending``, ``jobs:leased``, a specific 

13``job:<jb>`` hash) are passed in as KEYS by the Python layer via redis_keys. 

14``claim_pending`` is the sole exception — it discovers a job id by RPOP and so 

15must build that job's ``job:<jb>`` hash key and its 

16``submission:<sid>:current_job`` pointer key from inside Lua; those two literal 

17templates mirror ``redis_keys.job`` / ``redis_keys.submission_current_job`` and 

18are the only place key names are spelled outside redis_keys. 

19 

20Corrupted state (partial Redis data loss — never produced by normal operation) 

21is reaped lazily, in passing: the job hash is the single source of truth and 

22the pending list / leased set are only indexes, so a script that stumbles on a 

23dangling entry destroys or unindexes it and moves on instead of erroring. The 

24affected submission stays Pending and is rescued by rejudge (spec §12). 

25""" 

26 

27from dataclasses import dataclass 

28from typing import Any 

29 

30# claim_pending — pending → leased, with dispatch-side currency destroy (INV4). 

31# 

32# KEYS[1] = jobs:pending (LIST) KEYS[2] = jobs:leased (SET) 

33# ARGV[1] = now ARGV[2] = runner_id ARGV[3] = lease_ttl_sec 

34# 

35# Loop popping the FIFO tail: a job whose hash has evaporated or lost its 

36# submission_id (currency would be unverifiable — corrupted beyond judging) is 

37# destroyed and skipped; a job that is no longer its submission's current_job is 

38# destroyed on the spot (the dispatch-side half of INV4 — a superseded rejudge 

39# job never runs); the first live current job is leased (attempts+1, 

40# state=leased) and its id returned. A missing attempts field falls back to 0 

41# instead of erroring — the id is already popped at that point, so a script 

42# error would silently lose the job. Empty queue → nil. 

43CLAIM_PENDING = ''' 

44local now = tonumber(ARGV[1]) 

45local runner = ARGV[2] 

46local ttl = tonumber(ARGV[3]) 

47while true do 

48 local jb = redis.call('RPOP', KEYS[1]) 

49 if not jb then 

50 return nil 

51 end 

52 local job_key = 'job:' .. jb 

53 local sid = redis.call('HGET', job_key, 'submission_id') 

54 if not sid then 

55 -- Hash evaporated (DEL is a no-op then) or corrupted beyond judging (no 

56 -- submission_id means currency cannot be verified): destroy, move on. 

57 redis.call('DEL', job_key) 

58 else 

59 local current = redis.call('GET', 'submission:' .. sid .. ':current_job') 

60 if current == jb then 

61 local attempts = (tonumber(redis.call('HGET', job_key, 'attempts')) or 0) + 1 

62 redis.call('HSET', job_key, 

63 'attempts', attempts, 

64 'leased_by', runner, 

65 'lease_deadline', now + ttl, 

66 'state', 'leased') 

67 redis.call('SADD', KEYS[2], jb) 

68 return jb 

69 else 

70 redis.call('DEL', job_key) 

71 end 

72 end 

73end 

74''' 

75 

76# renew_lease — heartbeat extends an owned lease (spec §7.2). 

77# 

78# KEYS[1] = jobs:leased (SET) KEYS[2] = job:<jb> (HASH) 

79# ARGV[1] = job_id ARGV[2] = runner_id ARGV[3] = now ARGV[4] = lease_ttl_sec 

80# 

81# Only when the job is still leased AND owned by the caller: push the deadline to 

82# now+ttl, return 1. Anything else (not leased, wrong owner, evaporated hash) → 

83# 0. An evaporated hash additionally reaps the ghost member from jobs:leased — 

84# the hash is the source of truth, the set only an index. Never creates or 

85# resurrects a key (SREM only removes) — HSET runs only on the confirmed-owner 

86# path, where the hash provably exists. 

87RENEW_LEASE = ''' 

88if redis.call('SISMEMBER', KEYS[1], ARGV[1]) == 0 then 

89 return 0 

90end 

91if redis.call('EXISTS', KEYS[2]) == 0 then 

92 -- Hash is the source of truth; a member whose hash evaporated is a ghost -- 

93 -- reap it so the orphan scan stops rescanning it forever. 

94 redis.call('SREM', KEYS[1], ARGV[1]) 

95 return 0 

96end 

97local leased_by = redis.call('HGET', KEYS[2], 'leased_by') 

98if leased_by == ARGV[2] then 

99 redis.call('HSET', KEYS[2], 'lease_deadline', tonumber(ARGV[3]) + tonumber(ARGV[4])) 

100 return 1 

101end 

102return 0 

103''' 

104 

105# reclaim_expired — expired lease → new leaseholder, or converge (spec §7.3, §9). 

106# 

107# KEYS[1] = jobs:leased (SET) KEYS[2] = job:<jb> (HASH) 

108# ARGV[1] = job_id ARGV[2] = new_runner ARGV[3] = now 

109# ARGV[4] = lease_ttl_sec ARGV[5] = max_attempts 

110# 

111# Eligibility is decided ONLY by lease_deadline vs now (INV2) — runner liveness 

112# never enters. If expired and attempts < max: attempts+1, hand to new_runner, 

113# fresh deadline, return 1. If expired and attempts already >= max: return -1 and 

114# leave the job UNTOUCHED (INV5 boundary; the JE landing sweep is slice 3, not 

115# this script's job). Not leased / not expired → 0. An evaporated hash → 0 and 

116# the ghost member is reaped from jobs:leased (hash is the source of truth); a 

117# hash that still exists but lost its lease_deadline field stays tracked — a 

118# runner may well be judging it, and tearing down live state risks losing the 

119# job entirely. 

120RECLAIM_EXPIRED = ''' 

121if redis.call('SISMEMBER', KEYS[1], ARGV[1]) == 0 then 

122 return 0 

123end 

124if redis.call('EXISTS', KEYS[2]) == 0 then 

125 -- Hash is the source of truth; a member whose hash evaporated is a ghost -- 

126 -- reap it so the orphan scan stops rescanning it forever. 

127 redis.call('SREM', KEYS[1], ARGV[1]) 

128 return 0 

129end 

130local deadline = redis.call('HGET', KEYS[2], 'lease_deadline') 

131if not deadline then 

132 -- Hash alive but the field is gone: leave it tracked (see header note). 

133 return 0 

134end 

135local now = tonumber(ARGV[3]) 

136if tonumber(deadline) >= now then 

137 return 0 

138end 

139local attempts = tonumber(redis.call('HGET', KEYS[2], 'attempts')) 

140if attempts >= tonumber(ARGV[5]) then 

141 return -1 

142end 

143redis.call('HSET', KEYS[2], 

144 'attempts', attempts + 1, 

145 'leased_by', ARGV[2], 

146 'lease_deadline', now + tonumber(ARGV[4]), 

147 'state', 'leased') 

148return 1 

149''' 

150 

151 

152@dataclass(frozen=True) 

153class Scripts: 

154 claim_pending: Any 

155 renew_lease: Any 

156 reclaim_expired: Any 

157 

158 

159def load(client) -> Scripts: 

160 """Register the per-transition scripts on ``client`` and return them. 

161 

162 ``register_script`` is cheap (it only stores the body and computes the SHA 

163 lazily on first EVALSHA), so binding to the currently active shared client on 

164 each use keeps scripts coherent with the ``_cache`` reset done per test. 

165 """ 

166 return Scripts( 

167 claim_pending=client.register_script(CLAIM_PENDING), 

168 renew_lease=client.register_script(RENEW_LEASE), 

169 reclaim_expired=client.register_script(RECLAIM_EXPIRED), 

170 )