diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 5b3a7adf55b..50864005e10 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -357,10 +357,13 @@ jobs:
 
   postgres:
     runs-on: ubuntu-latest
-    needs: ioredis-redis
+    needs: [changes, ioredis-redis]
 
     name: testing node@${{ matrix.node-version }}, postgres@${{ matrix.postgres-version }}
 
+    env:
+      should_run: ${{ needs.changes.outputs.node == 'true' || needs.changes.outputs.lua == 'true' }}
+
     strategy:
       matrix:
         node-version: [lts/*]
@@ -386,18 +389,26 @@ jobs:
           --health-retries 5
 
     steps:
+      - name: Skip when node/lua inputs are unchanged
+        if: env.should_run != 'true'
+        run: echo "No node/lua changes detected; skipping postgres test steps."
       - name: Checkout repository
+        if: env.should_run == 'true'
         uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
       - name: Use Node.js ${{ matrix.node-version }}
+        if: env.should_run == 'true'
         uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0
         with:
           node-version: ${{ matrix.node-version }}
           cache: 'yarn'
-      - run: yarn install --ignore-engines --frozen-lockfile --non-interactive
+      - if: env.should_run == 'true'
+        run: yarn install --ignore-engines --frozen-lockfile --non-interactive
       # `yarn build` runs the pretest codegen (raw scripts, command transforms)
       # and compiles the sources the shared test classes import.
-      - run: yarn build
-      - run: yarn test:postgres
+      - if: env.should_run == 'true'
+        run: yarn build
+      - if: env.should_run == 'true'
+        run: yarn test:postgres
 
   bun-redis:
     runs-on: ubuntu-latest
diff --git a/python/bullmq/backend.py b/python/bullmq/backend.py
index dd578c448c3..994513c459d 100644
--- a/python/bullmq/backend.py
+++ b/python/bullmq/backend.py
@@ -348,6 +348,74 @@ async def waitForJob(self, block_timeout: float) -> Any:
         Returns the raw marker entry on success, or a falsy value on timeout.
         """
 
+    # ============================================================
+    # Job schedulers (repeatable job factories)
+    # ============================================================
+
+    @abstractmethod
+    async def addJobScheduler(
+        self,
+        job_scheduler_id: str,
+        next_millis: Optional[int],
+        template_data: str,
+        template_opts: dict,
+        scheduler_opts: dict,
+        delayed_job_opts: dict,
+        producer_id: Optional[str] = None,
+    ):
+        """Register/override a scheduler and enqueue its next iteration.
+
+        Returns a ``(job_id, delay)`` pair for the newly-scheduled iteration,
+        or a falsy value when no iteration was produced.
+        """
+
+    @abstractmethod
+    async def updateJobSchedulerNextMillis(
+        self,
+        job_scheduler_id: str,
+        next_millis: Optional[int],
+        template_data: str,
+        delayed_job_opts: dict,
+        producer_id: Optional[str] = None,
+    ):
+        """Advance an existing scheduler to its next iteration without
+        touching its template. Returns the new delayed job id, or a falsy
+        value if no iteration was produced (e.g. the scheduler is gone)."""
+
+    @abstractmethod
+    async def removeJobScheduler(self, job_scheduler_id: str) -> int:
+        """Remove a scheduler and its pending next-iteration job.
+
+        Returns 0 if the scheduler was removed, 1 if it did not exist.
+        """
+
+    @abstractmethod
+    async def isJobScheduler(self, job_scheduler_id: str) -> bool:
+        """Return whether ``job_scheduler_id`` is a registered scheduler."""
+
+    @abstractmethod
+    async def getJobScheduler(self, job_scheduler_id: str):
+        """Return a ``(fields, next_millis)`` pair for a single scheduler.
+
+        ``fields`` is the metadata mapping (``name``, ``ic``, ``every``,
+        ``pattern``, ``data``, ``opts`` ...) in the Redis-hash shape that
+        :func:`bullmq.job_scheduler._transform_scheduler_data` consumes, or a
+        falsy value when the scheduler is missing. ``next_millis`` is the
+        next-run timestamp, or ``None``.
+        """
+
+    @abstractmethod
+    async def getJobSchedulers(
+        self, start: int = 0, end: int = -1, asc: bool = False
+    ) -> list:
+        """Return a page of registered schedulers as a list of
+        ``(key, fields, next_millis)`` tuples (see :meth:`getJobScheduler`
+        for the ``fields`` shape)."""
+
+    @abstractmethod
+    async def getJobSchedulersCount(self) -> int:
+        """Return the total number of registered schedulers."""
+
 
 BackendFactory = Callable[..., Backend]
 """Factory that builds a :class:`Backend` for a given queue.
diff --git a/python/bullmq/backends/postgres_backend.py b/python/bullmq/backends/postgres_backend.py
index d5524b2d22a..e217d7c408f 100644
--- a/python/bullmq/backends/postgres_backend.py
+++ b/python/bullmq/backends/postgres_backend.py
@@ -15,6 +15,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import time
 from typing import Any, Optional, TYPE_CHECKING
@@ -165,6 +166,48 @@ def _normalize_keep(remove_on: Any) -> tuple[bool, Optional[int], Optional[int]]
     return (False, None, None)
 
 
+def _scheduler_row_to_hash(row: dict) -> tuple[dict, Optional[int]]:
+    """Map a ``scheduler`` row into the Redis-hash-shaped ``(fields, next)``
+    pair that :func:`bullmq.job_scheduler._transform_scheduler_data` consumes.
+
+    Scalar fields become string values; ``template_data``/``template_opts``
+    (stored as ``jsonb``, decoded by psycopg into Python objects) are
+    re-serialized to JSON strings and omitted when empty. Absent fields are
+    dropped. Mirrors the Node ``mapSchedulerRow`` helper.
+    """
+    fields: dict = {}
+
+    def _put(key: str, value: Any) -> None:
+        if value is not None:
+            fields[key] = str(value)
+
+    _put("name", row.get("name"))
+    _put("ic", row.get("iteration_count"))
+    _put("limit", row.get("limit_count"))
+    _put("startDate", row.get("start_date_ms"))
+    _put("endDate", row.get("end_date_ms"))
+    _put("tz", row.get("tz"))
+    _put("pattern", row.get("pattern"))
+    _put("every", row.get("every_ms"))
+    _put("offset", row.get("offset_ms"))
+
+    template_data = row.get("template_data")
+    if template_data is not None:
+        data = template_data if isinstance(template_data, str) else _json(template_data)
+        if data != "{}":
+            fields["data"] = data
+
+    template_opts = row.get("template_opts")
+    if template_opts is not None:
+        opts = template_opts if isinstance(template_opts, str) else _json(template_opts)
+        if opts != "{}":
+            fields["opts"] = opts
+
+    next_run = row.get("next_run_ms")
+    next_millis = None if next_run is None else _to_int(next_run)
+    return (fields, next_millis)
+
+
 class PostgresBackend(Backend):
     """PostgreSQL adapter implementing :class:`~bullmq.backend.Backend`."""
 
@@ -753,6 +796,122 @@ async def waitForJob(self, block_timeout: float) -> Any:
             if not checked_waiting_job and await self._has_waiting_job():
                 return marker
 
+    # ============================================================
+    # Job schedulers (repeatable job factories)
+    # ============================================================
+
+    async def addJobScheduler(
+        self,
+        job_scheduler_id: str,
+        next_millis: Optional[int],
+        template_data: str,
+        template_opts: dict,
+        scheduler_opts: dict,
+        delayed_job_opts: dict,
+        producer_id: Optional[str] = None,
+    ):
+        result = await self._run(
+            "add_job_scheduler",
+            [
+                self.queue_name,
+                job_scheduler_id,
+                next_millis,
+                template_data or "{}",
+                _jsonb(template_opts or {}),
+                _jsonb(scheduler_opts or {}),
+                _jsonb(delayed_job_opts or {}),
+                _now_ms(),
+                producer_id,
+            ],
+            op="addJobScheduler",
+        )
+        row = result.first_map()
+        if not row or row.get("job_id") is None:
+            return None
+        return (str(row["job_id"]), _to_int(row.get("delay")))
+
+    async def updateJobSchedulerNextMillis(
+        self,
+        job_scheduler_id: str,
+        next_millis: Optional[int],
+        template_data: str,
+        delayed_job_opts: dict,
+        producer_id: Optional[str] = None,
+    ):
+        result = await self._run(
+            "update_job_scheduler",
+            [
+                self.queue_name,
+                job_scheduler_id,
+                next_millis,
+                template_data or "{}",
+                _jsonb(delayed_job_opts or {}),
+                _now_ms(),
+                producer_id,
+            ],
+            op="updateJobSchedulerNextMillis",
+        )
+        row = result.first_map()
+        job_id = row.get("job_id") if row else None
+        return str(job_id) if job_id is not None else None
+
+    async def removeJobScheduler(self, job_scheduler_id: str) -> int:
+        result = await self._run(
+            "remove_job_scheduler", [self.queue_name, job_scheduler_id]
+        )
+        row = result.first_map()
+        return _to_int(row.get("removed")) if row else 1
+
+    async def isJobScheduler(self, job_scheduler_id: str) -> bool:
+        result = await self._run(
+            "is_job_scheduler", [self.queue_name, job_scheduler_id]
+        )
+        row = result.first_map()
+        return bool(row.get("exists")) if row else False
+
+    async def getJobScheduler(self, job_scheduler_id: str):
+        result = await self._run(
+            "get_job_scheduler", [self.queue_name, job_scheduler_id]
+        )
+        row = result.first_map()
+        if not row:
+            return (None, None)
+        return _scheduler_row_to_hash(row)
+
+    async def getJobSchedulers(
+        self, start: int = 0, end: int = -1, asc: bool = False
+    ) -> list:
+        count = None if end < 0 else end - start + 1
+        result = await self._run(
+            "get_job_schedulers_range", [self.queue_name, asc, start, count]
+        )
+        # The range command only returns (scheduler_id, next_run_ms); fetch
+        # each scheduler's metadata concurrently, mirroring the Redis path.
+        rows = result.maps()
+        if not rows:
+            return []
+        details = await asyncio.gather(
+            *(
+                self._run("get_job_scheduler", [self.queue_name, row["scheduler_id"]])
+                for row in rows
+            )
+        )
+        out = []
+        for row, detail in zip(rows, details):
+            detail_row = detail.first_map()
+            key = row["scheduler_id"]
+            if detail_row:
+                fields, next_millis = _scheduler_row_to_hash(detail_row)
+            else:
+                fields, next_millis = {}, _to_int(row.get("next_run_ms"))
+            out.append((key, fields, next_millis))
+        return out
+
+    async def getJobSchedulersCount(self) -> int:
+        result = await self._run("get_job_schedulers_count", [self.queue_name])
+        row = result.first_map()
+        return _to_int(row.get("count")) if row else 0
+
 
 def create_postgres_backend(
     name: str,
diff --git a/python/bullmq/backends/redis_backend.py b/python/bullmq/backends/redis_backend.py
index 67ca01b500d..0327fb1cc98 100644
--- a/python/bullmq/backends/redis_backend.py
+++ b/python/bullmq/backends/redis_backend.py
@@ -16,6 +16,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from typing import Any, Optional, TYPE_CHECKING
 
 from bullmq.backend import Backend
@@ -39,6 +40,20 @@
 )
 
 
+def _array_to_dict(arr) -> dict:
+    """Turn the Lua-flat ``[k1, v1, k2, v2, ...]`` shape into a dict.
+    ``redis-py`` may also already hand us a dict (depending on the response
+    policy); pass it through unchanged in that case."""
+    if isinstance(arr, dict):
+        return arr
+    if not arr:
+        return {}
+    out = {}
+    for i in range(0, len(arr), 2):
+        out[arr[i]] = arr[i + 1]
+    return out
+
+
 class RedisBackend(Backend):
     """Redis adapter implementing :class:`~bullmq.backend.Backend`."""
 
@@ -406,6 +421,95 @@ async def removeDeprecatedPriorityKey(self) -> Any:
     async def waitForJob(self, block_timeout: float) -> Any:
         return await self.bclient.bzpopmin(self.keys["marker"], block_timeout)
 
+    # ============================================================
+    # Job schedulers (repeatable job factories)
+    # ============================================================
+
+    async def addJobScheduler(
+        self,
+        job_scheduler_id: str,
+        next_millis: Optional[int],
+        template_data: str,
+        template_opts: dict,
+        scheduler_opts: dict,
+        delayed_job_opts: dict,
+        producer_id: Optional[str] = None,
+    ):
+        return await self.scripts.addJobScheduler(
+            job_scheduler_id,
+            next_millis,
+            template_data,
+            template_opts,
+            scheduler_opts,
+            delayed_job_opts,
+            producer_id,
+        )
+
+    async def updateJobSchedulerNextMillis(
+        self,
+        job_scheduler_id: str,
+        next_millis: Optional[int],
+        template_data: str,
+        delayed_job_opts: dict,
+        producer_id: Optional[str] = None,
+    ):
+        return await self.scripts.updateJobSchedulerNextMillis(
+            job_scheduler_id,
+            next_millis,
+            template_data,
+            delayed_job_opts,
+            producer_id,
+        )
+
+    async def removeJobScheduler(self, job_scheduler_id: str) -> int:
+        return await self.scripts.removeJobScheduler(job_scheduler_id)
+
+    async def isJobScheduler(self, job_scheduler_id: str) -> bool:
+        # Probe the `ic` (iteration count) field on the per-id hash so that
+        # legacy repeatable-job ids sharing the `repeat` sorted set are not
+        # misclassified as schedulers.
+        scheduler_hash_key = f"{self.keys['repeat']}:{job_scheduler_id}"
+        exists = await self.conn.hexists(scheduler_hash_key, "ic")
+        return exists == 1
+
+    async def getJobScheduler(self, job_scheduler_id: str):
+        raw, score = await self.scripts.getJobScheduler(job_scheduler_id)
+        next_millis = int(score) if score is not None else None
+        return (_array_to_dict(raw) if raw else None, next_millis)
+
+    async def getJobSchedulers(
+        self, start: int = 0, end: int = -1, asc: bool = False
+    ) -> list:
+        repeat_key = self.keys["repeat"]
+        if asc:
+            raw = await self.conn.zrange(repeat_key, start, end, withscores=True)
+        else:
+            raw = await self.conn.zrevrange(repeat_key, start, end, withscores=True)
+
+        if not raw:
+            return []
+
+        members = [member for member, _score in raw]
+        scores = [score for _member, score in raw]
+
+        # Issue the per-scheduler HGETALLs concurrently to avoid an N+1
+        # sequential round-trip on large scheduler counts.
+        fields_per_member = await asyncio.gather(
+            *(self.conn.hgetall(f"{repeat_key}:{member}") for member in members)
+        )
+
+        out = []
+        for member, score, fields in zip(members, scores, fields_per_member):
+            try:
+                next_millis = int(score)
+            except (TypeError, ValueError):
+                next_millis = None
+            out.append((member, fields, next_millis))
+        return out
+
+    async def getJobSchedulersCount(self) -> int:
+        return await self.conn.zcard(self.keys["repeat"])
+
 
 def create_redis_backend(
     name: str,
diff --git a/python/bullmq/job_scheduler.py b/python/bullmq/job_scheduler.py
index 81aeab5a319..ff2569f50bf 100644
--- a/python/bullmq/job_scheduler.py
+++ b/python/bullmq/job_scheduler.py
@@ -23,7 +23,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import json
 import time
 from datetime import datetime, timezone
@@ -122,7 +121,9 @@ class JobScheduler:
 
     def __init__(self, queue: "Queue", repeat_strategy=None):
         self.queue = queue
-        self.scripts = queue.scripts
+        # All datastore operations go through the queue's backend, so the
+        # scheduler is backend-agnostic (Redis, PostgreSQL, ...).
+        self.backend = queue.backend
         self.repeat_strategy = repeat_strategy or default_repeat_strategy
 
     async def upsertJobScheduler(
@@ -230,7 +231,7 @@ async def upsertJobScheduler(
                 k: v for k, v in scheduler_opts.items() if v is not None
             }
 
-            result = await self.scripts.addJobScheduler(
+            result = await self.backend.addJobScheduler(
                 job_scheduler_id,
                 clamped_next,
                 template_data_str,
@@ -257,7 +258,7 @@ async def upsertJobScheduler(
             return job
 
         # Non-override path: only advance the next-millis pointer.
-        job_id = await self.scripts.updateJobSchedulerNextMillis(
+        job_id = await self.backend.updateJobSchedulerNextMillis(
             job_scheduler_id,
             next_millis or now,
             template_data_str,
@@ -310,89 +311,36 @@ def _build_next_job_opts(
 
     async def removeJobScheduler(self, job_scheduler_id: str) -> int:
         """Remove a scheduler. Returns 0 on success, 1 if absent."""
-        return await self.scripts.removeJobScheduler(job_scheduler_id)
+        return await self.backend.removeJobScheduler(job_scheduler_id)
 
     async def isJobScheduler(self, job_scheduler_id: str) -> bool:
-        """
-        Return True if `job_scheduler_id` corresponds to a registered
-        scheduler. Probes the `ic` field on the per-id hash so that
-        legacy repeatable-job ids stored in the same sorted set are not
-        misclassified as schedulers. Mirrors Node's `isJobScheduler`.
-        """
-        scheduler_hash_key = f"{self.queue.keys['repeat']}:{job_scheduler_id}"
-        exists = await self.queue.client.hexists(scheduler_hash_key, "ic")
-        return exists == 1
+        """Return True if `job_scheduler_id` corresponds to a registered
+        scheduler. Mirrors Node's `isJobScheduler`."""
+        return await self.backend.isJobScheduler(job_scheduler_id)
 
     async def getScheduler(self, job_scheduler_id: str) -> Optional[dict]:
         """Return the JSON-shaped scheduler record, or None."""
-        raw, score = await self.scripts.getJobScheduler(job_scheduler_id)
-        next_millis = int(score) if score is not None else None
-        if not raw:
+        fields, next_millis = await self.backend.getJobScheduler(job_scheduler_id)
+        if not fields:
             return None
-        fields = _array_to_dict(raw)
         return _transform_scheduler_data(job_scheduler_id, fields, next_millis)
 
     async def getJobSchedulers(
         self, start: int = 0, end: int = -1, asc: bool = False
     ) -> list:
         """Page through registered schedulers. `asc=True` returns
-        earliest-next-fire first.
-
-        Issues the per-scheduler `HGETALL` calls concurrently via
-        `asyncio.gather` to avoid an N+1 sequential round-trip on large
-        scheduler counts.
-        """
-        repeat_key = self.queue.keys["repeat"]
-        if asc:
-            raw = await self.queue.client.zrange(
-                repeat_key, start, end, withscores=True
-            )
-        else:
-            raw = await self.queue.client.zrevrange(
-                repeat_key, start, end, withscores=True
-            )
-
-        if not raw:
-            return []
-
-        members = [member for member, _score in raw]
-        scores = [score for _member, score in raw]
-
-        fields_per_member = await asyncio.gather(
-            *(
-                self.queue.client.hgetall(f"{repeat_key}:{member}")
-                for member in members
-            )
-        )
-
+        earliest-next-fire first."""
+        records = await self.backend.getJobSchedulers(start, end, asc)
         out = []
-        for member, score, fields_raw in zip(members, scores, fields_per_member):
-            try:
-                next_millis = int(score)
-            except (TypeError, ValueError):
-                next_millis = None
-            data = _transform_scheduler_data(member, fields_raw, next_millis)
+        for key, fields, next_millis in records:
+            data = _transform_scheduler_data(key, fields, next_millis)
             if data is not None:
                 out.append(data)
         return out
 
     async def getSchedulersCount(self) -> int:
         """Total number of registered schedulers."""
-        return await self.queue.client.zcard(self.queue.keys["repeat"])
-
-
-def _array_to_dict(arr) -> dict:
-    """Turn the Lua-flat `[k1, v1, k2, v2, ...]` shape into a dict.
-    `redis-py` may also already hand us a dict (depending on the
-    response policy); pass it through unchanged in that case."""
-    if isinstance(arr, dict):
-        return arr
-    if not arr:
-        return {}
-    out = {}
-    for i in range(0, len(arr), 2):
-        out[arr[i]] = arr[i + 1]
-    return out
+        return await self.backend.getJobSchedulersCount()
 
 
 def _transform_scheduler_data(
diff --git a/python/bullmq/queue.py b/python/bullmq/queue.py
index 8e2575f97c6..f82211ae597 100644
--- a/python/bullmq/queue.py
+++ b/python/bullmq/queue.py
@@ -474,10 +474,6 @@ def jobScheduler(self):
         on first use so that queues which never schedule pay no cost.
         """
         if self._job_scheduler is None:
-            if self.scripts is None or self.client is None:
-                raise NotImplementedError(
-                    "Job schedulers are currently only supported by the Redis backend"
-                )
             from bullmq.job_scheduler import JobScheduler
             self._job_scheduler = JobScheduler(self)
         return self._job_scheduler
diff --git a/python/bullmq/worker.py b/python/bullmq/worker.py
index 54f61273054..bc3523a76b2 100644
--- a/python/bullmq/worker.py
+++ b/python/bullmq/worker.py
@@ -121,6 +121,7 @@ def __init__(self, name: str, processor: Callable[..., asyncio.Future], opts: Wo
         self.client = getattr(self.backend, "conn", None)
         self.bclient = getattr(self.backend, "bclient", None)
         self.scripts = getattr(self.backend, "scripts", None)
+        self.keys = getattr(self.backend, "keys", None)
         self.prefix = self.opts.get("prefix", "bull")
         self.closing = False
         self.forceClosing = False
@@ -136,6 +137,7 @@ def __init__(self, name: str, processor: Callable[..., asyncio.Future], opts: Wo
         self.drained = False
         self.qualifiedName = self.backend.qualifiedName
         self.workerName = opts.get("name")
+        self._job_scheduler = None
         self.clientName = self.backend.clientName(
             f":w:{self.workerName}" if self.workerName else ""
         )
@@ -254,9 +256,22 @@ async def moveToActive(self, token: str):
         if result:
             job_data, id, limit_until, delay_until = result
 
-        return self.nextJobFromJobData(job_data, id, limit_until, delay_until, token)
+        return await self.nextJobFromJobData(job_data, id, limit_until, delay_until, token)
 
-    def nextJobFromJobData(self, job_data: dict | None = None, job_id: str | None = None, limit_until: int = 0,
+    @property
+    def jobScheduler(self):
+        """
+        Lazily-instantiated JobScheduler that shares this worker's backend
+        (same queue). Created on first use so that workers which never
+        process scheduled jobs pay no cost. Mirrors the Node worker's
+        `jobScheduler` getter.
+        """
+        if self._job_scheduler is None:
+            from bullmq.job_scheduler import JobScheduler
+            self._job_scheduler = JobScheduler(self)
+        return self._job_scheduler
+
+    async def nextJobFromJobData(self, job_data: dict | None = None, job_id: str | None = None, limit_until: int = 0,
         delay_until: int = 0, token: str | None = None) -> Job | None:
         self.limitUntil = max(limit_until, 0) or 0
 
@@ -272,8 +287,48 @@ def nextJobFromJobData(self, job_data: dict | None = None, job_id: str | None =
             self.drained = False
             job_instance = Job.fromJSON(self, job_data, job_id)
             job_instance.token = token
+
+            # If this job was produced by a job scheduler, advance the
+            # scheduler to materialize its next iteration. The Node worker
+            # performs the same step here in `nextJobFromJobData`; without
+            # it a scheduler only ever fires its first iteration
+            # (see issue #4483).
+            if job_instance.repeatJobKey:
+                try:
+                    await self.retryIfFailed(
+                        lambda: self._scheduleNextIteration(job_instance),
+                        {"delay_in_ms": self.opts.get("runRetryDelay")},
+                    )
+                except Exception as err:
+                    # Emit the error but don't propagate it: the current job
+                    # has already been moved to active and must still be
+                    # returned for processing. The trade-off is that the
+                    # next iteration will not have been scheduled.
+                    self.emit(
+                        "error",
+                        RuntimeError(
+                            "Failed to add repeatable job for next iteration: "
+                            f"{err}"
+                        ),
+                    )
+
             return job_instance
 
+    async def _scheduleNextIteration(self, job: Job) -> None:
+        """Upsert the job scheduler that produced `job` so its next
+        iteration is materialized. Most of the arguments are no longer
+        strictly needed (the scheduler reads them from its own record),
+        but they are passed through to mirror the Node implementation."""
+        await self.jobScheduler.upsertJobScheduler(
+            job.repeatJobKey,
+            (job.opts or {}).get("repeat"),
+            job.name,
+            job.data,
+            job.opts,
+            override=False,
+            producer_id=job.id,
+        )
+
     async def waitForJob(self) -> int:
         block_timeout = self.getBlockTimeout(self.blockUntil)
         block_timeout = block_timeout if self.backend.capabilities.get("canDoubleTimeout", False) else math.ceil(block_timeout)
diff --git a/python/tests/job_scheduler_test.py b/python/tests/job_scheduler_test.py
index a2e861643b7..069b08284b6 100644
--- a/python/tests/job_scheduler_test.py
+++ b/python/tests/job_scheduler_test.py
@@ -3,6 +3,7 @@
 """
 
 import os
+import asyncio
 import time
 import unittest
 from uuid import uuid4
@@ -11,7 +12,7 @@
 from croniter import CroniterBadCronError
 import redis.asyncio as redis
 
-from bullmq import Queue
+from bullmq import Queue, Worker
 from bullmq.job_scheduler import default_repeat_strategy, _transform_scheduler_data
 
 
@@ -70,6 +71,103 @@ async def test_upsert_pattern_creates_scheduler(self):
         finally:
             await queue.close()
 
+    async def test_worker_lazily_creates_job_scheduler(self):
+        """The worker exposes a lazily-instantiated `jobScheduler` that
+        shares its backend (same queue). It must be created on first use
+        and memoized thereafter."""
+        worker = Worker(self.queueName, None, {"prefix": prefix})
+        try:
+            self.assertIsNone(worker._job_scheduler)
+            scheduler = worker.jobScheduler
+            self.assertIsNotNone(scheduler)
+            # Memoized: repeated access returns the same instance.
+            self.assertIs(worker.jobScheduler, scheduler)
+            # It shares the worker's backend rather than opening its own.
+            self.assertIs(scheduler.queue, worker)
+            self.assertIs(scheduler.backend, worker.backend)
+        finally:
+            await worker.close()
+
+    async def test_worker_advances_scheduler_across_iterations(self):
+        """Regression test for #4483: a worker processing a job produced
+        by a job scheduler must advance the scheduler to its next
+        iteration. Previously only the first iteration was ever
+        materialized and the schedule stopped permanently."""
+        queue = Queue(self.queueName, {"prefix": prefix})
+        processed: list[str] = []
+        enough = asyncio.Event()
+
+        async def process(job, token=None):
+            processed.append(job.id)
+            if len(processed) >= 3:
+                enough.set()
+            return None
+
+        worker = None
+        try:
+            await queue.upsertJobScheduler(
+                "tick", {"every": 200}, job_name="tick"
+            )
+
+            worker = Worker(self.queueName, process, {"prefix": prefix})
+
+            # Wait until the scheduler has advanced past its first iteration
+            # (or bail out after a generous timeout to avoid hanging CI).
+            try:
+                await asyncio.wait_for(enough.wait(), timeout=10)
+            except asyncio.TimeoutError:
+                pass
+
+            # More than the single first iteration was processed.
+            self.assertGreaterEqual(len(processed), 3)
+            # Each processed job is a distinct scheduler iteration.
+            self.assertEqual(len(set(processed)), len(processed))
+            for job_id in processed:
+                self.assertTrue(job_id.startswith("repeat:tick:"))
+
+            # The scheduler is still registered and its iteration count
+            # advanced beyond the initial upsert.
+            scheduler = await queue.getJobScheduler("tick")
+            self.assertIsNotNone(scheduler)
+            self.assertGreater(scheduler["iterationCount"], 1)
+
+            # A delayed job representing the next iteration is always pending.
+            self.assertEqual(await queue.getDelayedCount(), 1)
+        finally:
+            if worker is not None:
+                await worker.close()
+            await queue.close()
+
+    async def test_next_job_from_job_data_materializes_next_iteration(self):
+        """`Worker.nextJobFromJobData` must upsert the scheduler for the
+        next iteration when the job carries a `repeatJobKey`, mirroring the
+        Node worker. Exercised directly (no timing) for determinism."""
+        queue = Queue(self.queueName, {"prefix": prefix})
+        worker = Worker(self.queueName, None, {"prefix": prefix})
+        try:
+            first = await queue.upsertJobScheduler(
+                "direct", {"every": 5_000}, job_name="direct"
+            )
+            self.assertIsNotNone(first)
+
+            # Grab the raw stored job data for the first iteration and feed
+            # it back through the worker exactly like the run loop would.
+            job_data = await worker.backend.getJobData(first.id)
+            self.assertTrue(job_data)
+
+            next_job = await worker.nextJobFromJobData(
+                job_data, first.id, token="token"
+            )
+            self.assertIsNotNone(next_job)
+            self.assertEqual(next_job.id, first.id)
+
+            # The scheduler advanced to its second iteration.
+            scheduler = await queue.getJobScheduler("direct")
+            self.assertEqual(scheduler["iterationCount"], 2)
+        finally:
+            await worker.close()
+            await queue.close()
+
     async def test_upsert_override_replaces_pending_iteration(self):
         queue = Queue(self.queueName, {"prefix": prefix})
         try: