# Workflow With Cron Schedule Not pulled by new workers on same task queue

**URL:** <https://community.temporal.io/t/workflow-with-cron-schedule-not-pulled-by-new-workers-on-same-task-queue/13452>\
**Category:** Community Support\
**Tags:** python-sdk\
**Created:** [September 6, 2024, 7:28pm UTC](https://community.temporal.io/t/workflow-with-cron-schedule-not-pulled-by-new-workers-on-same-task-queue/13452 "2024-09-06T19:28:32Z")\
**Posts on this page:** 10\
**Page:** 1

<div class="post-metadata">

**Author:** ![Abdelrahman\_Abouzei1](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/abdelrahman_abouzei1/32/5824_2.png) [@Abdelrahman\_Abouzei1](https://community.temporal.io/u/Abdelrahman_Abouzei1)\
**Post date:** [September 6, 2024, 7:28pm UTC](https://community.temporal.io/t/workflow-with-cron-schedule-not-pulled-by-new-workers-on-same-task-queue/13452/1 "2024-09-06T19:28:32Z")

</div>

Hello, I will attach the context below

1. I have a corn job that’s scheduled to run every 1 minute defined as following

```auto
@workflow.defn(name="TaskExecutor")
class TaskExecutor:
    @workflow.run
    async def run(self) -> None:
        return await workflow.execute_activity(
            Activities.PollExecuteTasks,
            start_to_close_timeout=timedelta(days=365),
        )

caller:
def initTaskExecutor():
    print('INITTASKEXECUTOR called init_tast_executor')
    result = asyncio.run(settings.TEMPORAL_CLIENT.start_workflow(
        Workflows.TaskExecutor.run,
        id="workflow-process-task-queue-new",
        task_queue="task-queue-1",
        id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE,
        cron_schedule="* * * * *",
    ))
    return result

```

1. Whenever the worker that called this dies and another is spawned it also tries to start it but gets error: `Workflow execution already started`
2. Here is my worker code:

```auto
async def main(): 
    client = settings.TEMPORAL_CLIENT

    with concurrent.futures.ThreadPoolExecutor(max_workers=100) as activity_executor:
        worker = Worker(
            client,
            task_queue="task-queue-1",
            workflows=[
                ... 
                Workflows.TaskExecutor, 
            ],
            activities=[
                ...
            ],
            activity_executor=activity_executor,
        )
        print("Starting the worker....")
        await worker.run()

if __name__ == " __main__":
    while True:
        try:
            asyncio.run(main())
        except Exception as e:
            # Log the exception
            print(f"Exception occurred: {e}")
            print("Retrying in 5 seconds...")

```

The issue is that despite the workflow still running on Temporal cloud it never gets scheduled to the new worker. (It is scheduled properly, if I terminated the corn job and let the new worker start the workflow again)

Thank you in advance.

---

<div class="post-metadata">

**Author:** ![Abdelrahman\_Abouzei1](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/abdelrahman_abouzei1/32/5824_2.png) [@Abdelrahman\_Abouzei1](https://community.temporal.io/u/Abdelrahman_Abouzei1)\
**Post date:** [September 8, 2024, 3:32pm UTC](https://community.temporal.io/t/workflow-with-cron-schedule-not-pulled-by-new-workers-on-same-task-queue/13452/2 "2024-09-08T15:32:22Z")

</div>

@maxim Any input out be appreciated, thank you!

---

<div class="post-metadata">

**Author:** ![maxim](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/maxim/32/8_2.png) [@maxim](https://community.temporal.io/u/maxim)\
**Post date:** [September 8, 2024, 6:27pm UTC](https://community.temporal.io/t/workflow-with-cron-schedule-not-pulled-by-new-workers-on-same-task-queue/13452/3 "2024-09-08T18:27:05Z")

</div>

Do you handle the “Workflow execution already started” exception? Could it be that the worker was not appropriately started when this exception was thrown?

---

<div class="post-metadata">

**Author:** ![Abdelrahman\_Abouzei1](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/abdelrahman_abouzei1/32/5824_2.png) [@Abdelrahman\_Abouzei1](https://community.temporal.io/u/Abdelrahman_Abouzei1)\
**Post date:** [September 8, 2024, 6:47pm UTC](https://community.temporal.io/t/workflow-with-cron-schedule-not-pulled-by-new-workers-on-same-task-queue/13452/4 "2024-09-08T18:47:29Z")

</div>

I don’t think, I would go to the [cloud.temporal.io](http://cloud.temporal.io) dashboard and I would see the number of workers reflecting the workers running.

Here is my call stack, I run it as a django Management command as following: `Python3 manage.py run_worker`

```auto
import asyncio
from django.core.management.base import BaseCommand
from plato_app.workflows import worker as Worker

class Command(BaseCommand):
    help = 'Runs the Temporal worker'

    def add_arguments(self, parser):
        # Optional: add command line arguments here
        pass

    def handle(self, *args, **options):
        # Your script logic here
        print('RUN_WORKER HANDLE about to run worker.main()')
        asyncio.run(Worker.main())
        self.stdout.write(self.style.SUCCESS('Successfully ran script'))

```

Worker:

```auto
async def main(): #StevenTodo: client below may be relevant to some changes I made earlier to settings.py temporal client
    client = settings.TEMPORAL_CLIENT

    with concurrent.futures.ThreadPoolExecutor(max_workers=100) as activity_executor:
        worker = Worker(
            client,
            task_queue="task-queue-1",
            workflows=[
                Workflows.YourWorkflow, 
                Workflows.TaskExecutor, 
                Workflows.AddAudioClassFile, 
                Workflows.AddPDFClassFile, 
                Workflows.WorkflowCaller
            ],
            activities=[
                Activities.your_activity, 
                Activities.PollExecuteTasks, 
                Activities.FileSpeechToText, 
                Activities.ChunkVectorizeSrtFile, 
                Activities.FilePDFToText,
                Activities.ChunkVectorizeTextFile,
                Activities.MarkProcessedWF,
            ],
            activity_executor=activity_executor,
        )
        print("Starting the worker....")
        await worker.run()

if __name__ == " __main__":
    while True:
        try:
            asyncio.run(main())
        except Exception as e:
            # Log the exception
            print(f"Exception occurred: {e}")
            print("Retrying in 5 seconds...")

```

---

<div class="post-metadata">

**Author:** ![maxim](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/maxim/32/8_2.png) [@maxim](https://community.temporal.io/u/maxim)\
**Post date:** [September 8, 2024, 7:24pm UTC](https://community.temporal.io/t/workflow-with-cron-schedule-not-pulled-by-new-workers-on-same-task-queue/13452/5 "2024-09-08T19:24:17Z")

</div>

I don’t know anything about djungo to know if a command is the right approach here.

What is the workflow history in Temporal UI/CLI?

---

<div class="post-metadata">

**Author:** ![Abdelrahman\_Abouzei1](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/abdelrahman_abouzei1/32/5824_2.png) [@Abdelrahman\_Abouzei1](https://community.temporal.io/u/Abdelrahman_Abouzei1)\
**Post date:** [September 12, 2024, 3:04pm UTC](https://community.temporal.io/t/workflow-with-cron-schedule-not-pulled-by-new-workers-on-same-task-queue/13452/6 "2024-09-12T15:04:31Z")

</div>

@maxim Here is the history.  
Also, I realized if I redeploy the same code as is, it works fine and pulls the scheduled code otherwise it doesn’t.

```auto
{
  "eventId": "1",
  "eventTime": "2024-09-12T03:52:01.353488198Z",
  "eventType": "WorkflowExecutionStarted",
  "version": "1301",
  "taskId": "104876508",
  "workflowExecutionStartedEventAttributes": {
    "workflowType": {
      "name": "TaskExecutor"
    },
    "taskQueue": {
      "name": "task-queue-1",
      "kind": "TASK_QUEUE_KIND_NORMAL"
    },
    "workflowTaskTimeout": "10s",
    "continuedExecutionRunId": "765cfa3e-3a7f-4635-8a1b-8dfcc3a56970",
    "initiator": "CONTINUE_AS_NEW_INITIATOR_CRON_SCHEDULE",
    "lastCompletionResult": {
      "payloads": [
        null
      ]
    },
    "originalExecutionRunId": "c3564772-61fe-40e5-a98d-fdab1a02fe8c",
    "firstExecutionRunId": "ad8456cb-9937-42ee-9049-7206c9a0def7",
    "attempt": 1,
    "cronSchedule": "* * * * *",
    "firstWorkflowTaskBackoff": "59s",
    "prevAutoResetPoints": {
      "points": [
        {
          "binaryChecksum": "3ac9fa34ebcbe2c4bc2313589bf438cd",
          "runId": "ad8456cb-9937-42ee-9049-7206c9a0def7",
          "firstWorkflowTaskCompletedId": "4",
          "createTime": "2024-09-07T02:48:01.070633067Z",
          "expireTime": "2024-10-07T02:48:04.691561946Z",
          "resettable": true
        }
      ]
    },
    "workflowId": "workflow-process-task-queue-new"
  },
  "links": []
}

```

---

<div class="post-metadata">

**Author:** ![maxim](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/maxim/32/8_2.png) [@maxim](https://community.temporal.io/u/maxim)\
**Post date:** [September 12, 2024, 6:12pm UTC](https://community.temporal.io/t/workflow-with-cron-schedule-not-pulled-by-new-workers-on-same-task-queue/13452/7 "2024-09-12T18:12:58Z")

</div>

Is this the complete history consisting of a single event?

---

<div class="post-metadata">

**Author:** ![Abdelrahman\_Abouzei1](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/abdelrahman_abouzei1/32/5824_2.png) [@Abdelrahman\_Abouzei1](https://community.temporal.io/u/Abdelrahman_Abouzei1)\
**Post date:** [September 12, 2024, 11:33pm UTC](https://community.temporal.io/t/workflow-with-cron-schedule-not-pulled-by-new-workers-on-same-task-queue/13452/8 "2024-09-12T23:33:24Z")

</div>

That’s the History of the Cron Scheduled job on Temporal UI. All of it yes (At least all of what I can see in the History section of the UI)

---

<div class="post-metadata">

**Author:** ![tihomir](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/tihomir/32/6580_2.png) [@tihomir](https://community.temporal.io/u/tihomir)\
**Post date:** [September 13, 2024, 2:19am UTC](https://community.temporal.io/t/workflow-with-cron-schedule-not-pulled-by-new-workers-on-same-task-queue/13452/9 "2024-09-13T02:19:51Z")

</div>

Could you create a support ticket ([support.temporal.io](http://support.temporal.io)) where please put your namespace name (so you don’t share it here publicly) and then share your support ticket number here in thread and will take a look.

Does this happen all the time or just for that workflow execution?  
Can you try to signalWithStart the running cron execution from a client and see if that triggers scheduling of first workflow task?

---

<div class="post-metadata">

**Author:** ![Abdelrahman\_Abouzei1](https://sea2.discourse-cdn.com/flex016/user_avatar/community.temporal.io/abdelrahman_abouzei1/32/5824_2.png) [@Abdelrahman\_Abouzei1](https://community.temporal.io/u/Abdelrahman_Abouzei1)\
**Post date:** [September 13, 2024, 10:35pm UTC](https://community.temporal.io/t/workflow-with-cron-schedule-not-pulled-by-new-workers-on-same-task-queue/13452/10 "2024-09-13T22:35:13Z")

</div>

Ticket number: 8329

I tried running it as follows:

```auto
    await client.start_workflow(
        Workflows.TaskExecutor.run, 
        id="workflow-process-task-queue-new",
        task_queue="task-queue-1",
        start_signal="submit_greeting",
        start_signal_args=["User Signal with Start"],
    )

```

Still no worker is running it. However other workflows get scheduled normally when the worker is updated.

Here is how I define this workflow compared to other workflows, and here is how I am running the worker:

```auto
from datetime import timedelta
import pickle
from temporalio import workflow
from temporalio.common import WorkflowIDReusePolicy
from temporalio.workflow import ParentClosePolicy

# Import our activity, passing it through the sandbox
with workflow.unsafe.imports_passed_through():
    from plato_app.workflows import wf_params as WfParams, activities as Activities, activity_params as ActivityParams
    

# ---------------------------------- Test Activities ---------------------------------- 
@workflow.defn(name="YourWorkflow")
class YourWorkflow:
    @workflow.run
    async def run(self, params: WfParams.YourParams) -> str:
        return await workflow.execute_activity(
            Activities.your_activity,
            params,
            start_to_close_timeout=timedelta(seconds=10),
        )

# ---------------------------------- Taskk Executor Workflows ---------------------------------- 
@workflow.defn(name="TaskExecutor")
class TaskExecutor:
    @workflow.run
    async def run(self) -> None:
        return await workflow.execute_activity(
            Activities.PollExecuteTasks,
            start_to_close_timeout=timedelta(days=365),
        )
    
@workflow.defn(name="WorkFlowCaller")
class WorkflowCaller:
    @workflow.run
    async def run(self, params: WfParams.WorkFlowCallerParams) -> None:
        from .events import EVENTS_TO_WORKFLOWS, EVENTS_TO_WORKFLOW_ID_PREFIX

        print("Executing and wiating on this wf: ", params.workflow_id, "event_type: ", params.event_type)
        wf_params = pickle.loads(params.payload)
        print('WF_PARAMS is ', wf_params)
        _ = await workflow.execute_child_workflow(
            EVENTS_TO_WORKFLOWS[params.event_type].run,
            wf_params,
            id=EVENTS_TO_WORKFLOW_ID_PREFIX[params.event_type] + params.workflow_id,
            task_queue="task-queue-1",
            id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE_FAILED_ONLY,
            parent_close_policy=ParentClosePolicy.ABANDON
        )
        
        print("\nwf executed successfully, marking as processed: ", params.workflow_id, "event_type: ", params.event_type)
        return await workflow.execute_activity(
            Activities.MarkProcessedWF,
            ActivityParams.MarkProcessedWFParams(
                workflow_id=params.workflow_id
            ),
            start_to_close_timeout=timedelta(seconds=100),
        )

```

That’s how I run the worker:

```auto
import asyncio
import concurrent.futures
from temporalio.worker import Worker
from . import workflows as Workflows
from . import activities as Activities
from django.conf import settings
from temporalio.worker import Worker
from plato_app.workflows import workflows as Workflows
from plato_app.workflows import activities as Activities

async def main(): 
    client = settings.TEMPORAL_CLIENT

    with concurrent.futures.ThreadPoolExecutor(max_workers=100) as activity_executor:
        worker = Worker(
            client,
            task_queue="task-queue-1",
            workflows=[
                Workflows.YourWorkflow, 
                Workflows.TaskExecutor, 
                Workflows.AddAudioClassFile, 
                Workflows.AddPDFClassFile, 
                Workflows.WorkflowCaller
            ],
            activities=[
                Activities.your_activity, 
                Activities.PollExecuteTasks, 
                Activities.FileSpeechToText, 
                Activities.ChunkVectorizeSrtFile, 
                Activities.FilePDFToText,
                Activities.ChunkVectorizeTextFile,
                Activities.MarkProcessedWF,
            ],
            activity_executor=activity_executor,
        )
        print("Starting the worker....")
        await worker.run()

if __name__ == " __main__":
    while True:
        try:
            asyncio.run(main())
        except Exception as e:
            # Log the exception
            print(f"Exception occurred: {e}")
            print("Retrying in 5 seconds...")

```