fix(locked_tasks): Add timeout to auto unlock tasks #16

gagantrivedi · 2025-01-06T05:15:27Z

Fixes #17 by picking up locked tasks for processing after a timeout.

gagantrivedi · 2025-01-06T08:37:10Z

task_processor/decorators.py

@@ -161,6 +167,7 @@ def register_recurring_task(
    args: tuple[typing.Any] = (),
    kwargs: dict[str, typing.Any] | None = None,
    first_run_time: time | None = None,
+    timeout: timedelta | None = timedelta(minutes=30),


The longest run duration for a recurring task in production is 00:11:49.801789

gagantrivedi · 2025-01-06T08:38:28Z

task_processor/migrations/0012_add_locked_at_and_timeout.py

+        migrations.AddField(
+            model_name="task",
+            name="timeout",
+            field=models.DurationField(blank=True, null=True),


Default not set here to avoid locking the table.

gagantrivedi · 2025-01-06T08:40:38Z

task_processor/decorators.py

@@ -124,6 +128,7 @@ def register_task_handler(  # noqa: C901
    queue_size: int | None = None,
    priority: TaskPriority = TaskPriority.NORMAL,
    transaction_on_commit: bool = True,
+    timeout: timedelta | None = timedelta(seconds=60),


The longest run duration for a task in production is 00:00:30.459713

gagantrivedi · 2025-01-06T08:46:20Z

task_processor/migrations/sql/0012_get_recurringtasks_to_process.sql

@@ -0,0 +1,32 @@
+CREATE OR REPLACE FUNCTION get_recurringtasks_to_process(num_tasks integer)


Only lines 10, 20, and 25 were updated here.

task_processor/processor.py

matthewelwell

Added a few fairly minor comments / questions, but overall this looks excellent 👌

matthewelwell · 2025-01-07T09:44:40Z

task_processor/models.py

@@ -146,6 +151,9 @@ def mark_success(self):
 class RecurringTask(AbstractBaseTask):
    run_every = models.DurationField()
    first_run_time = models.TimeField(blank=True, null=True)
+    locked_at = models.DateTimeField(blank=True, null=True)


I guess we don't need this on the regular Task model because we can use scheduled_for ?

I initially planned to add the unlock feature only for recurring tasks, but on second thought, I have now added it to regular tasks as well.

task_processor/migrations/0012_add_locked_at_and_timeout.py

task_processor/migrations/sql/0012_get_recurringtasks_to_process.sql

tests/unit/task_processor/test_unit_task_processor_processor.py

matthewelwell · 2025-01-07T10:19:11Z

tests/unit/task_processor/test_unit_task_processor_processor.py

+    task_runs = run_recurring_tasks()
+
+    # Then
+    assert cache.get(DEFAULT_CACHE_KEY)


Suggested change

assert cache.get(DEFAULT_CACHE_KEY)

assert cache.get(DEFAULT_CACHE_KEY) == DEFAULT_CACHE_VALUE

I do also think it's worth adding an assert to the beginning of the test that DEFAULT_CACHE_KEY doesn't already exist in the cache, or a cache.delete(DEFAULT_CACHE_KEY) perhaps?

matthewelwell · 2025-01-08T11:45:14Z

tests/unit/task_processor/test_unit_task_processor_processor.py

@@ -63,6 +70,116 @@ def test_run_task_runs_task_and_creates_task_run_object_when_success(db):
    assert task.completed


+def test_run_tasks_runs_locked_task_after_tiemout(


Suggested change

def test_run_tasks_runs_locked_task_after_tiemout(

def test_run_tasks_runs_locked_task_after_timeout(

matthewelwell · 2025-01-08T11:47:56Z

task_processor/migrations/sql/0012_get_tasks_to_process.sql

+        WHERE id = row_to_return.id;
+        -- If we don't explicitly update the columns here, the client will receive a row
+        -- that is locked but still shows `is_locked` as `False` and `locked_at` as `None`.
+        row_to_return.is_locked := TRUE;


Should this also do row_to_return.locked_at = NOW() (or maybe NOW() should be stored in a variable for exact accuracy)?

matthewelwell · 2025-01-08T11:48:40Z

task_processor/migrations/sql/0012_get_recurringtasks_to_process.sql

+        -- If we don't explicitly update the columns here, the client will receive a row
+        -- that is locked but still shows `is_locked` as `False` and `locked_at` as `None`.
+        row_to_return.is_locked := TRUE;
+        row_to_return.locked_at := NOW();


nitpicking, but we should set NOW() to some variable for exact accuracy here, right?

No, NOW() returns the same value on every call in a transaction

gagantrivedi force-pushed the fix/locked-tasks branch 3 times, most recently from 28e82e6 to dc63661 Compare January 6, 2025 08:13

gagantrivedi commented Jan 6, 2025

View reviewed changes

gagantrivedi force-pushed the fix/locked-tasks branch from dc63661 to cba7ead Compare January 6, 2025 08:39

gagantrivedi commented Jan 6, 2025

View reviewed changes

gagantrivedi force-pushed the fix/locked-tasks branch from cba7ead to f0ac380 Compare January 6, 2025 08:43

gagantrivedi changed the title ~~fix(locked-tasks): wip~~ fix(17/recurring-task-lock): Add timeout to auto unlock task Jan 6, 2025

gagantrivedi linked an issue Jan 6, 2025 that may be closed by this pull request

Recurring Tasks can get stuck in locked status #17

Open

gagantrivedi commented Jan 6, 2025

View reviewed changes

task_processor/processor.py Show resolved Hide resolved

gagantrivedi force-pushed the fix/locked-tasks branch 2 times, most recently from dfd51cd to 6e07161 Compare January 6, 2025 08:55

fix(17/recurring-task-lock): Add timeout to auto unlock task

6e07161

gagantrivedi marked this pull request as ready for review January 6, 2025 10:14

gagantrivedi requested review from a team and matthewelwell and removed request for a team January 6, 2025 10:14

matthewelwell reviewed Jan 7, 2025

View reviewed changes

gagantrivedi force-pushed the fix/locked-tasks branch from ed052fa to d8a4c37 Compare January 8, 2025 04:23

gagantrivedi changed the title ~~fix(17/recurring-task-lock): Add timeout to auto unlock task~~ fix(locked_tasks): Add timeout to auto unlock tasks Jan 8, 2025

fix locks for non recurring tasks

d8a4c37

gagantrivedi requested a review from matthewelwell January 8, 2025 04:26

matthewelwell approved these changes Jan 8, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix(locked_tasks): Add timeout to auto unlock tasks #16

fix(locked_tasks): Add timeout to auto unlock tasks #16

gagantrivedi commented Jan 6, 2025 •

edited

Loading

gagantrivedi Jan 6, 2025 •

edited

Loading

gagantrivedi Jan 6, 2025

gagantrivedi Jan 6, 2025

gagantrivedi Jan 6, 2025

matthewelwell left a comment •

edited

Loading

matthewelwell Jan 7, 2025

gagantrivedi Jan 8, 2025

matthewelwell Jan 7, 2025

matthewelwell Jan 7, 2025

gagantrivedi Jan 8, 2025

matthewelwell Jan 8, 2025

matthewelwell Jan 8, 2025

matthewelwell Jan 8, 2025

gagantrivedi Jan 8, 2025

		@@ -0,0 +1,32 @@
		CREATE OR REPLACE FUNCTION get_recurringtasks_to_process(num_tasks integer)

	assert cache.get(DEFAULT_CACHE_KEY)
	assert cache.get(DEFAULT_CACHE_KEY) == DEFAULT_CACHE_VALUE

		@@ -63,6 +70,116 @@ def test_run_task_runs_task_and_creates_task_run_object_when_success(db):
		assert task.completed


		def test_run_tasks_runs_locked_task_after_tiemout(

fix(locked_tasks): Add timeout to auto unlock tasks #16

Are you sure you want to change the base?

fix(locked_tasks): Add timeout to auto unlock tasks #16

Conversation

gagantrivedi commented Jan 6, 2025 • edited Loading

gagantrivedi Jan 6, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

matthewelwell left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gagantrivedi commented Jan 6, 2025 •

edited

Loading

gagantrivedi Jan 6, 2025 •

edited

Loading

matthewelwell left a comment •

edited

Loading