update the json atomically

Author: nsaje

tasktiger/redis_scripts.py

@@ -261,12 +261,15 @@
 """
 
 # KEYS = { scheduled_zset_key, task_data_key }
-# ARGV = { score, member, serialized_task_data }
+# ARGV = { score, member }
+# score is used both as the new ZSET score and as the new scheduled_at value.
 UPDATE_SCHEDULED_TIME = """
     local score = redis.call('zscore', KEYS[1], ARGV[2])
     if score then
         redis.call('zadd', KEYS[1], ARGV[1], ARGV[2])
-        redis.call('set', KEYS[2], ARGV[3])
+        local data = cjson.decode(redis.call('get', KEYS[2]))
+        data['scheduled_at'] = tonumber(ARGV[1])
+        redis.call('set', KEYS[2], cjson.encode(data))
         return 1
     else
         return 0
@@ -587,16 +590,15 @@ def update_scheduled_time(
         task_data_key: str,
         score: float,
         member: str,
-        serialized_task_data: str,
     ) -> bool:
         """
-        Atomically updates a task's scheduled time in the ZSET and its _data
-        blob. Returns True if the task was found and updated, False if the task
-        was not present in the ZSET.
+        Atomically updates a task's scheduled time in the ZSET and patches
+        scheduled_at in the _data blob. Returns True if the task was found and
+        updated, False if the task was not present in the ZSET.
         """
         result = self._update_scheduled_time(
             keys=[scheduled_zset_key, task_data_key],
-            args=[score, member, serialized_task_data],
+            args=[score, member],
         )
         return bool(result)
 

tasktiger/task.py

@@ -431,13 +431,11 @@ def update_scheduled_time(
         ts = get_timestamp(when)
         assert ts
 
-        new_data = {**self._data, "scheduled_at": ts}
         found = tiger.scripts.update_scheduled_time(
             scheduled_zset_key=tiger._key(SCHEDULED, self.queue),
             task_data_key=tiger._key("task", self.id),
             score=ts,
             member=self.id,
-            serialized_task_data=json.dumps(new_data),
         )
         if not found:
             raise TaskNotFound(
@@ -446,7 +444,7 @@ def update_scheduled_time(
                 )
             )
 
-        self._data = new_data
+        self._data["scheduled_at"] = ts
         self._ts = ts
 
     def __repr__(self) -> str: