Skip to content

[ENG-401] fix: token queue should respect sub queues while calling serve next #3708

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nandkishorr wants to merge 7 commits into
base: develop
Choose a base branch
from
Open

[ENG-401] fix: token queue should respect sub queues while calling serve next #3708

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
637e5ce
fix:added proper token assigning
nandkishorr Jul 10, 2026
9d383a7
fix:added proper token validation and clear current token
nandkishorr Jul 10, 2026
bc2eec6
Merge branch 'develop' into ENG-401-token-queue-should-respect-sub-qu…
nandkishorr Jul 10, 2026
14ac7ae
greptile review and fix tests
nandkishorr Jul 10, 2026
a7ace39
Merge branch 'ENG-401-token-queue-should-respect-sub-queues-while-cal…
nandkishorr Jul 10, 2026
140b0b1
cleanup
nandkishorr Jul 10, 2026
b21fa4a
added lock
nandkishorr Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 46 additions & 29 deletions care/emr/api/viewsets/scheduling/token.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,17 +81,21 @@ def perform_create(self, instance):
super().perform_create(instance)

def validate_data(self, instance, model_obj=None):
if (
model_obj
and instance.sub_queue
and model_obj.sub_queue
and instance.sub_queue != model_obj.sub_queue.external_id
):
existing_current = TokenSubQueue.objects.filter(
current_token=model_obj
).exists()
if existing_current:
raise ValidationError("Sub Queue already has a current token")
if model_obj and instance.sub_queue:
if (
model_obj.sub_queue
and instance.sub_queue != model_obj.sub_queue.external_id
):
existing_current = TokenSubQueue.objects.filter(
current_token=model_obj
).exists()
if existing_current:
raise ValidationError("Sub Queue already has a current token")
if (
instance.sub_queue
and instance.status == TokenStatusOptions.IN_PROGRESS.value
):
raise ValidationError("Use set_next endpoint to assign a token.")

return super().validate_data(instance, model_obj)
Comment on lines 83 to 100

@greptile-apps greptile-apps Bot Jul 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 IN_PROGRESS guard bypassed when sub_queue is null

The new validation at lines 94–98 that blocks direct IN_PROGRESS updates lives inside if model_obj and instance.sub_queue:. A caller that sends {"status": "in_progress", "sub_queue": null} satisfies pydantic (since TokenUpdateSpec.sub_queue: UUID4 | None accepts null) but makes instance.sub_queue falsy, so the entire guard is skipped. The token then transitions to IN_PROGRESS with sub_queue = null in perform_update — the old sub-queue's current_token gets cleared, but nothing sets a new current_token, leaving the queue in an inconsistent state.

The IN_PROGRESS status check should be guarded on model_obj alone, independent of whether a sub-queue is being provided in the same request.


Expand All @@ -100,18 +104,26 @@ def perform_update(self, instance):
raise ValidationError("Sub Queue and Queue are not in the same facility")
with transaction.atomic():
obj = self.get_object()
if obj.sub_queue and obj.sub_queue != instance.sub_queue:
if (
instance.sub_queue
and obj.sub_queue.resource != instance.sub_queue.resource
):
raise ValidationError(
"Sub Queue and Queue are not in the same resource"
)
# Clear current token if the sub queue is changed
if obj.sub_queue.current_token == obj:
obj.sub_queue.current_token = None
obj.sub_queue.save(update_fields=["current_token", "modified_date"])
if (
obj.sub_queue
and obj.sub_queue != instance.sub_queue
and instance.sub_queue
and obj.sub_queue.resource != instance.sub_queue.resource
):
raise ValidationError(
"Sub Queue and Queue are not in the same resource"
)
# Clear current token if the sub queue is changed
if (
obj.sub_queue
and obj.sub_queue.current_token == obj
and (
obj.sub_queue != instance.sub_queue
or instance.status != TokenStatusOptions.IN_PROGRESS.value
)
):
obj.sub_queue.current_token = None
obj.sub_queue.save(update_fields=["current_token", "modified_date"])
super().perform_update(instance)

def perform_destroy(self, instance):
Expand Down Expand Up @@ -181,17 +193,22 @@ def get_queryset(self):
@action(detail=True, methods=["POST"])
def set_next(self, request, *args, **kwargs):
obj = self.get_object()
if obj.status != TokenStatusOptions.CREATED.value:
raise ValidationError("Token in serving state cannot be set next")
request_obj = SetCurrentTokenRequest(**request.data)
queue = obj.queue
self.authorize_update(None, obj)
with transaction.atomic():
sub_queue = get_object_or_404(
TokenSubQueue,
external_id=request_obj.sub_queue,
resource=queue.resource,
)
sub_queue = get_object_or_404(
TokenSubQueue,
external_id=request_obj.sub_queue,
resource=queue.resource,
)
with Lock(f"token:set_next:{sub_queue.id}"), transaction.atomic():
if sub_queue.current_token and sub_queue.current_token != obj:
raise ValidationError("Sub Queue already has a current token")
sub_queue.current_token = obj
sub_queue.save()
obj.status = TokenStatusOptions.IN_PROGRESS.value
obj.sub_queue = sub_queue
obj.save()
Comment on lines +201 to 213

@greptile-apps greptile-apps Bot Jul 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Stale sub_queue snapshot inside the lock

sub_queue is fetched via get_object_or_404 (lines 201–205) before the lock is acquired. The lock serialises concurrent calls, but it does not refresh the in-memory object. A second request that fetches sub_queue (seeing current_token = None) and then acquires the lock after the first request has already committed will still evaluate sub_queue.current_token against the old cached value — both pass the guard and both tokens become IN_PROGRESS on the same sub-queue simultaneously.

Re-fetch sub_queue with select_for_update() inside the lock+transaction so the guard always reflects the current DB state:

with Lock(f"token:set_next:{sub_queue.id}"), transaction.atomic():
    sub_queue = TokenSubQueue.objects.select_for_update().get(id=sub_queue.id)
    if sub_queue.current_token and sub_queue.current_token != obj:
        raise ValidationError("Sub Queue already has a current token")
    ...

return Response(self.get_retrieve_pydantic_model().serialize(obj).to_json())
7 changes: 4 additions & 3 deletions care/emr/api/viewsets/scheduling/token_queue.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,9 +232,10 @@ def set_next_token_to_subqueue(self, request, *args, **kwargs):
).order_by("created_date")
if category:
tokens_qs = tokens_qs.filter(category=category)
if tokens_qs.exists():
next_token = tokens_qs.first()
else:
tokens_in_waiting_qs = tokens_qs.filter(sub_queue__isnull=True)
tokens_in_calling_qs = tokens_qs.filter(sub_queue=sub_queue)
next_token = tokens_in_calling_qs.first() or tokens_in_waiting_qs.first()
if next_token is None:
raise ValidationError("No tokens found")
Comment thread
nandkishorr marked this conversation as resolved.
sub_queue.current_token = next_token
sub_queue.save()
Expand Down
10 changes: 5 additions & 5 deletions care/emr/tests/test_token_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -275,7 +275,7 @@ def test_update_token_as_superuser(self):
name="Sub Queue 1",
)
token_data = {
"status": TokenStatusOptions.IN_PROGRESS,
"status": TokenStatusOptions.CREATED,
"note": "Token is in progress",
"sub_queue": subqueue.external_id,
}
Expand All @@ -289,7 +289,7 @@ def test_update_token_as_superuser(self):
format="json",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["status"], TokenStatusOptions.IN_PROGRESS.value)
self.assertEqual(response.data["status"], TokenStatusOptions.CREATED.value)

def test_update_token_as_user_with_permission(self):
"""Test updating a token as a user with permission."""
Expand All @@ -311,7 +311,7 @@ def test_update_token_as_user_with_permission(self):
name="Sub Queue 1",
)
token_data = {
"status": TokenStatusOptions.IN_PROGRESS,
"status": TokenStatusOptions.CREATED,
"note": "Token is in progress",
"sub_queue": subqueue.external_id,
}
Expand All @@ -325,7 +325,7 @@ def test_update_token_as_user_with_permission(self):
format="json",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["status"], TokenStatusOptions.IN_PROGRESS.value)
self.assertEqual(response.data["status"], TokenStatusOptions.CREATED.value)

def test_update_token_as_user_without_permission(self):
"""Test updating a token as a user without permission."""
Expand All @@ -342,7 +342,7 @@ def test_update_token_as_user_without_permission(self):
name="Sub Queue 1",
)
token_data = {
"status": TokenStatusOptions.IN_PROGRESS,
"status": TokenStatusOptions.CREATED,
"note": "Token is in progress",
"sub_queue": subqueue.external_id,
}
Expand Down
Loading