refactor(cache): refresh redis submodule
zhenyi committed
Modified lib/cache/redis/mod.rs lib/cache/redis/mod.rs +120 −2 View lib/cache/redis/mod.rs
@@ -23,6 +23,20 @@
2323 return 1
2424"#;
2525
26const COMPARE_AND_DELETE_SCRIPT: &str = r#"
27 if redis.call('GET', KEYS[1]) ~= ARGV[1] then
28 return 0
29 end
30 return redis.call('DEL', KEYS[1])
31"#;
32
33const REFRESH_LOCK_SCRIPT: &str = r#"
34 if redis.call('GET', KEYS[1]) ~= ARGV[1] then
35 return 0
36 end
37 return redis.call('EXPIRE', KEYS[1], ARGV[2])
38"#;
39
2640#[derive(Debug, thiserror::Error)]
2741pub enum RedisError {
2842 #[error("redis error: {0}")]
@@ -189,12 +203,112 @@
189203 Ok(())
190204 }
191205
206 /// Atomically set a key only if it does not already exist (`SET NX EX`).
207 /// Used for distributed locks. Returns `true` when the lock was acquired.
208 pub async fn set_nx(
209 &self,
210 key: &str,
211 value: Vec<u8>,
212 ttl_seconds: u64,
213 ) -> Result<bool, RedisError> {
214 let full_key = self.prefixed(key);
215 let mut conn = self.spawn_conn();
216 let set: Option<()> = redis::cmd("SET")
217 .arg(&*full_key)
218 .arg(value)
219 .arg("NX")
220 .arg("EX")
221 .arg(ttl_seconds as i64)
222 .query_async(&mut conn)
223 .await?;
224 Ok(set.is_some())
225 }
226
227 /// Atomically delete a key only if it currently holds the expected value
228 /// (Lua GET + DEL). Returns `true` when a value was deleted.
229 pub async fn compare_and_delete(&self, key: &str, expected: &[u8]) -> Result<bool, RedisError> {
230 let full_key = self.prefixed(key);
231 let mut conn = self.spawn_conn();
232 let deleted: i64 = redis::Script::new(COMPARE_AND_DELETE_SCRIPT)
233 .key(full_key)
234 .arg(expected)
235 .invoke_async(&mut conn)
236 .await?;
237 Ok(deleted == 1)
238 }
239
240 /// Atomically extend the TTL of a distributed lock only if the key still
241 /// holds the expected value (Lua GET + EXPIRE). Returns `true` when the
242 /// lock was refreshed.
243 pub async fn refresh_lock(
244 &self,
245 key: &str,
246 value: &[u8],
247 ttl_seconds: u64,
248 ) -> Result<bool, RedisError> {
249 let full_key = self.prefixed(key);
250 let mut conn = self.spawn_conn();
251 let refreshed: i64 = redis::Script::new(REFRESH_LOCK_SCRIPT)
252 .key(full_key)
253 .arg(value)
254 .arg(ttl_seconds as i64)
255 .invoke_async(&mut conn)
256 .await?;
257 Ok(refreshed == 1)
258 }
259
260 /// Push a value onto the tail of a list (`RPUSH`). Returns the new list length.
261 pub async fn rpush(&self, key: &str, value: Vec<u8>) -> Result<u64, RedisError> {
262 let full_key = self.prefixed(key);
263 let mut conn = self.spawn_conn();
264 let len: u64 = conn.rpush(&*full_key, value).await?;
265 Ok(len)
266 }
267
268 /// Push a value onto the head of a list (`LPUSH`). Returns the new list length.
269 pub async fn lpush(&self, key: &str, value: Vec<u8>) -> Result<u64, RedisError> {
270 let full_key = self.prefixed(key);
271 let mut conn = self.spawn_conn();
272 let len: u64 = conn.lpush(&*full_key, value).await?;
273 Ok(len)
274 }
275
276 /// Pop and return the head of a list (`LPOP`), if any.
277 pub async fn lpop(&self, key: &str) -> Result<Option<Vec<u8>>, RedisError> {
278 let full_key = self.prefixed(key);
279 let mut conn = self.spawn_conn();
280 Ok(redis::cmd("LPOP")
281 .arg(&*full_key)
282 .query_async(&mut conn)
283 .await?)
284 }
285
286 /// Remove up to `count` occurrences of `member` from a list (`LREM`).
287 /// Returns the number of removed elements.
288 pub async fn lrem(&self, key: &str, count: i64, member: &[u8]) -> Result<u64, RedisError> {
289 let full_key = self.prefixed(key);
290 let mut conn = self.spawn_conn();
291 let removed: u64 = conn.lrem(&*full_key, count as isize, member).await?;
292 Ok(removed)
293 }
294
295 /// Return the length of a list (`LLEN`).
296 pub async fn llen(&self, key: &str) -> Result<u64, RedisError> {
297 let full_key = self.prefixed(key);
298 let mut conn = self.spawn_conn();
299 let len: u64 = conn.llen(&*full_key).await?;
300 Ok(len)
301 }
302
192303 /// Atomically fetch-and-delete a key (GETDEL). Returns the previous value
193304 /// if present, enabling single-use consumption of one-time tokens.
194305 pub async fn getdel_bytes(&self, key: &str) -> Result<Option<Vec<u8>>, RedisError> {
195306 let full_key = self.prefixed(key);
196307 let mut conn = self.spawn_conn();
197 let value: Option<Vec<u8>> = redis::cmd("GETDEL").arg(&*full_key).query_async(&mut conn).await?;
308 let value: Option<Vec<u8>> = redis::cmd("GETDEL")
309 .arg(&*full_key)
310 .query_async(&mut conn)
311 .await?;
198312 Ok(value)
199313 }
200314
@@ -276,7 +390,11 @@
276390 Ok(())
277391 }
278392
279 pub async fn zremrange_by_score_lt(&self, key: &str, max_score: f64) -> Result<u64, RedisError> {
393 pub async fn zremrange_by_score_lt(
394 &self,
395 key: &str,
396 max_score: f64,
397 ) -> Result<u64, RedisError> {
280398 let full_key = self.prefixed(key);
281399 let mut conn = self.spawn_conn();
282400 let removed: u64 = conn.zrembyscore(&*full_key, "-inf", max_score).await?;