fix(redis): store EXAT/PXAT options and fix cancellation race across pipelines

Author: datlechin

Plugins/RedisDriverPlugin/RedisCommandParser.swift

@@ -73,6 +73,8 @@ enum RedisOperation {
 struct RedisSetOptions {
     var ex: Int?
     var px: Int?
+    var exat: Int?
+    var pxat: Int?
     var nx: Bool = false
     var xx: Bool = false
 }
@@ -998,18 +1000,20 @@ struct RedisCommandParser {
                 guard i + 1 < args.count else {
                     throw RedisParseError.missingArgument("EXAT requires a value")
                 }
-                guard Int(args[i + 1]) != nil else {
+                guard let timestamp = Int(args[i + 1]) else {
                     throw RedisParseError.invalidArgument("EXAT value must be a positive integer")
                 }
+                options.exat = timestamp
                 hasOption = true
                 i += 1
             case "PXAT":
                 guard i + 1 < args.count else {
                     throw RedisParseError.missingArgument("PXAT requires a value")
                 }
-                guard Int(args[i + 1]) != nil else {
+                guard let timestamp = Int(args[i + 1]) else {
                     throw RedisParseError.invalidArgument("PXAT value must be a positive integer")
                 }
+                options.pxat = timestamp
                 hasOption = true
                 i += 1
             case "NX":

Plugins/RedisDriverPlugin/RedisPluginConnection.swift

@@ -331,7 +331,7 @@ final class RedisPluginConnection: @unchecked Sendable {
         }
     }
 
-    private func resetCancellation() {
+    func resetCancellation() {
         stateLock.lock()
         _isCancelled = false
         stateLock.unlock()
@@ -356,7 +356,6 @@ final class RedisPluginConnection: @unchecked Sendable {
     func executeCommand(_ args: [String]) async throws -> RedisReply {
         #if canImport(CRedis)
         return try await pluginDispatchAsync(on: queue) { [self] in
-            resetCancellation()
             guard !isShuttingDown else {
                 throw RedisPluginError.notConnected
             }
@@ -379,7 +378,6 @@ final class RedisPluginConnection: @unchecked Sendable {
     func executePipeline(_ commands: [[String]]) async throws -> [RedisReply] {
         #if canImport(CRedis)
         return try await pluginDispatchAsync(on: queue) { [self] in
-            resetCancellation()
             guard !isShuttingDown else {
                 throw RedisPluginError.notConnected
             }
@@ -404,7 +402,6 @@ final class RedisPluginConnection: @unchecked Sendable {
     func selectDatabase(_ index: Int) async throws {
         #if canImport(CRedis)
         try await pluginDispatchAsync(on: queue) { [self] in
-            resetCancellation()
             guard !isShuttingDown else {
                 throw RedisPluginError.notConnected
             }

Plugins/RedisDriverPlugin/RedisPluginDriver.swift

@@ -78,6 +78,7 @@ final class RedisPluginDriver: PluginDatabaseDriver, @unchecked Sendable {
         let startTime = Date()
         cachedScanPattern = nil
         cachedScanKeys = nil
+        redisConnection?.resetCancellation()
 
         guard let conn = redisConnection else {
             throw RedisPluginError.notConnected
@@ -610,6 +611,8 @@ private extension RedisPluginDriver {
             if let opts = options {
                 if let ex = opts.ex { args += ["EX", String(ex)] }
                 if let px = opts.px { args += ["PX", String(px)] }
+                if let exat = opts.exat { args += ["EXAT", String(exat)] }
+                if let pxat = opts.pxat { args += ["PXAT", String(pxat)] }
                 if opts.nx { args.append("NX") }
                 if opts.xx { args.append("XX") }
             }