fix(tasks) Improve scheduler metrics and fix an error (#95960)

In debugging why the scheduler in s4s occasionally stops scheduling
tasks, I found some gaps in our metrics and logging that would have
helped solve the problem, and also a sentry error that occurs when the
'next' runtime for task is `now`. Redis doesn't allow `ex=0`.

Fixes SENTRY-FOR-SENTRY-B8D

Author: markstory

src/sentry/taskworker/scheduler/runner.py

@@ -46,7 +46,8 @@ def set(self, taskname: str, next_runtime: datetime) -> bool:
         Returns False when the key is set and a task should not be spawned.
         """
         now = timezone.now()
-        duration = next_runtime - now
+        # next_runtime & now could be the same second, and redis gets sad if ex=0
+        duration = max(int((next_runtime - now).total_seconds()), 1)
 
         result = self._redis.set(self._make_key(taskname), now.isoformat(), ex=duration, nx=True)
         return bool(result)
@@ -59,6 +60,8 @@ def read(self, taskname: str) -> datetime | None:
         result = self._redis.get(self._make_key(taskname))
         if result:
             return datetime.fromisoformat(result)
+
+        metrics.incr("taskworker.scheduler.run_storage.read.miss", tags={"taskname": taskname})
         return None
 
     def read_many(self, tasknames: list[str]) -> Mapping[str, datetime | None]:
@@ -203,6 +206,7 @@ def tick(self) -> float:
         self._update_heap()
 
         if not self._heap:
+            logger.warning("taskworker.scheduler.no_heap")
             return 60
 
         while True:
@@ -239,13 +243,22 @@ def _try_spawn(self, entry: ScheduleEntry) -> None:
                 },
             )
         else:
-            # sync with last_run state in storage
-            entry.set_last_run(self._run_storage.read(entry.fullname))
+            # We were not able to set a key, load last run from storage.
+            run_state = self._run_storage.read(entry.fullname)
+            entry.set_last_run(run_state)
 
-            logger.debug(
-                "taskworker.scheduler.sync_with_storage", extra={"fullname": entry.fullname}
+            logger.info(
+                "taskworker.scheduler.sync_with_storage",
+                extra={
+                    "taskname": entry.taskname,
+                    "namespace": entry.namespace,
+                    "last_runtime": run_state.isoformat() if run_state else None,
+                },
+            )
+            metrics.incr(
+                "taskworker.scheduler.sync_with_storage",
+                tags={"taskname": entry.taskname, "namespace": entry.namespace},
             )
-            metrics.incr("taskworker.scheduler.sync_with_storage")
 
     def _update_heap(self) -> None:
         """update the heap to reflect current remaining time"""
@@ -267,3 +280,10 @@ def _load_last_run(self) -> None:
         for item in self._entries:
             last_run = last_run_times.get(item.fullname, None)
             item.set_last_run(last_run)
+        logger.info(
+            "taskworker.scheduler.load_last_run",
+            extra={
+                "entry_count": len(self._entries),
+                "loaded_count": len(last_run_times),
+            },
+        )

tests/sentry/taskworker/scheduler/test_runner.py

@@ -34,6 +34,23 @@ def run_storage() -> RunStorage:
     return RunStorage(redis)
 
 
+def test_runstorage_zero_duration(run_storage: RunStorage) -> None:
+    with freeze_time("2025-07-19 14:25:00"):
+        now = timezone.now()
+        result = run_storage.set("test:do_stuff", now)
+        assert result is True
+
+
+def test_runstorage_double_set(run_storage: RunStorage) -> None:
+    with freeze_time("2025-07-19 14:25:00"):
+        now = timezone.now()
+        first = run_storage.set("test:do_stuff", now)
+        second = run_storage.set("test:do_stuff", now)
+
+        assert first is True, "initial set should return true"
+        assert second is False, "writing a key that exists should fail"
+
+
 @pytest.mark.django_db
 def test_schedulerunner_add_invalid(taskregistry) -> None:
     run_storage = Mock(spec=RunStorage)