From 843e11491509670bafe91d565d0b423cc897eeb7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Danii=C5=82=20M=2E?= Date: Sat, 26 Sep 2026 11:12:15 +0200 Subject: [PATCH 1/3] feat(gifts): prefix queue names and use sorted queue order strategy in Celery - Prefixes gift queues with 0-, 1-, and 2- for deterministic alphabetical sorting - Updates Celery queue_order_strategy to 'sorted' to avoid Kombu priority_cycle set-ordering bug - Enables task_acks_late to ensure workers fetch tasks just-in-time --- backend/core/src/core/constants.py | 6 +++--- backend/indexer_gifts/celery_app.py | 3 ++- docker-compose.yml | 2 +- 3 files changed, 6 insertions(+), 5 deletions(-) diff --git a/backend/core/src/core/constants.py b/backend/core/src/core/constants.py index 1aeffa2f..6e9ad1cc 100644 --- a/backend/core/src/core/constants.py +++ b/backend/core/src/core/constants.py @@ -44,10 +44,10 @@ CELERY_INDEX_PRICES_QUEUE_NAME = "index-prices-queue" # Gifts GIFT_COLLECTIONS_METADATA_KEY = "gifts-metadata-v3" -CELERY_GIFT_COLLECTIONS_SYNC_QUEUE = "gift-collections-sync" -CELERY_GIFT_USER_PRIORITY_QUEUE = "gift-user-priority" +CELERY_GIFT_COLLECTIONS_SYNC_QUEUE = "1-gift-collections-sync" +CELERY_GIFT_USER_PRIORITY_QUEUE = "0-gift-user-priority" USER_GIFT_REFRESH_COOLDOWN_SECONDS = 300 -CELERY_GIFT_USER_BATCH_QUEUE = "gift-user-batch" +CELERY_GIFT_USER_BATCH_QUEUE = "2-gift-user-batch" UPDATED_GIFT_USER_IDS = "updated_gift_user_ids" # ----------------- Paths --------------------- diff --git a/backend/indexer_gifts/celery_app.py b/backend/indexer_gifts/celery_app.py index aedaabba..76b435cc 100644 --- a/backend/indexer_gifts/celery_app.py +++ b/backend/indexer_gifts/celery_app.py @@ -16,7 +16,8 @@ def create_app() -> Celery: "result_expires": 300, # 5 minutes "include": ["indexer_gifts.tasks"], "worker_concurrency": gifts_indexer_settings.worker_concurrency, - "broker_transport_options": {"queue_order_strategy": "priority"}, + "broker_transport_options": {"queue_order_strategy": "sorted"}, + "task_acks_late": True, "worker_prefetch_multiplier": 1, } ) diff --git a/docker-compose.yml b/docker-compose.yml index 097974c7..3e64cfeb 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -114,7 +114,7 @@ services: "indexer_gifts.celery_app:app", "worker", "-Q", - "gift-user-priority,gift-collections-sync,gift-user-batch", + "0-gift-user-priority,1-gift-collections-sync,2-gift-user-batch", "--loglevel=info" ] environment: From fc92f879e92a22d635f4816417798bd0f2a4d96b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Danii=C5=82=20M=2E?= Date: Sat, 26 Sep 2026 12:16:14 +0200 Subject: [PATCH 2/3] feat(api): support timeout in wait_for_task and return 408 on async tasks endpoint --- backend/api/routes/system.py | 21 ++++++++++++++++++--- backend/core/src/core/constants.py | 1 + backend/core/src/core/utils/task.py | 6 ++++++ 3 files changed, 25 insertions(+), 3 deletions(-) diff --git a/backend/api/routes/system.py b/backend/api/routes/system.py index 6d1fa33d..fb6aa368 100644 --- a/backend/api/routes/system.py +++ b/backend/api/routes/system.py @@ -2,7 +2,11 @@ from fastapi import APIRouter, HTTPException, Depends from sqlalchemy.orm import Session -from starlette.status import HTTP_502_BAD_GATEWAY, HTTP_200_OK +from starlette.status import ( + HTTP_200_OK, + HTTP_408_REQUEST_TIMEOUT, + HTTP_502_BAD_GATEWAY, +) from api.deps import get_db_session from api.pos.base import BaseExceptionFDO @@ -19,16 +23,27 @@ "/async-tasks/{task_id}", responses={ HTTP_200_OK: {"model": StatusFDO}, + HTTP_408_REQUEST_TIMEOUT: { + "description": "Occurs when the task took longer than expected to complete", + "model": BaseExceptionFDO, + }, HTTP_502_BAD_GATEWAY: { "description": "Occurs when the system failed to complete the task", "model": BaseExceptionFDO, }, }, ) -async def get_task_status_status( +async def get_task_status( task_id: str, ) -> StatusFDO: - is_successful = await wait_for_task(task_id=task_id) + try: + is_successful = await wait_for_task(task_id=task_id) + except TimeoutError: + raise HTTPException( + detail="Task took longer than expected", + status_code=HTTP_408_REQUEST_TIMEOUT, + ) + if is_successful: return StatusFDO(status="success", message="Task is completed successfully") else: diff --git a/backend/core/src/core/constants.py b/backend/core/src/core/constants.py index 6e9ad1cc..1918457e 100644 --- a/backend/core/src/core/constants.py +++ b/backend/core/src/core/constants.py @@ -16,6 +16,7 @@ DEFAULT_CELERY_TASK_RETRY_DELAY = 60 DEFAULT_CELERY_TASK_MAX_RETRIES = 5 +DEFAULT_ASYNC_TASK_WAIT_TIMEOUT_SECONDS: float = 9.5 # Performance DEFAULT_BATCH_PROCESSING_SIZE = 5_000 diff --git a/backend/core/src/core/utils/task.py b/backend/core/src/core/utils/task.py index 0108ca0d..de55bd88 100644 --- a/backend/core/src/core/utils/task.py +++ b/backend/core/src/core/utils/task.py @@ -3,6 +3,7 @@ from celery import Celery from celery.result import AsyncResult +from core.constants import DEFAULT_ASYNC_TASK_WAIT_TIMEOUT_SECONDS from core.settings import core_settings @@ -23,12 +24,17 @@ async def wait_for_task( task_id: str | None = None, task_result: AsyncResult | None = None, interval: float = 1.0, + timeout: float = DEFAULT_ASYNC_TASK_WAIT_TIMEOUT_SECONDS, ) -> bool: if not task_result and task_id: task_result = AsyncResult(task_id) elif not task_result and not task_id: raise ValueError("Either task_id or task_result should be provided") + elapsed = 0.0 while not task_result.ready(): + if elapsed >= timeout: + raise TimeoutError("Task waiting timed out") await asyncio.sleep(interval) + elapsed += interval return task_result.successful() From ee0befc115fcd734dd96148a6ee076815c8bbb06 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Danii=C5=82=20M=2E?= Date: Sat, 26 Sep 2026 12:16:26 +0200 Subject: [PATCH 3/3] feat(fe): add timeout handling and feedback toasts for gifts refresh --- frontend/src/common/utils/constants.ts | 2 ++ .../ChatConditions/ChatConditions.tsx | 22 +++++++++++-- frontend/src/services/ApiService.ts | 31 +++++++++++++------ frontend/src/store/user/api.ts | 4 +++ frontend/src/store/user/user.ts | 13 +++++++- 5 files changed, 59 insertions(+), 13 deletions(-) diff --git a/frontend/src/common/utils/constants.ts b/frontend/src/common/utils/constants.ts index 300e4900..738f392a 100644 --- a/frontend/src/common/utils/constants.ts +++ b/frontend/src/common/utils/constants.ts @@ -4,6 +4,8 @@ export const API_VALIDATION_ERROR = 'Fill fields correctly' export const API_ERRORS = {} +export const ASYNC_TASK_TIMEOUT_MS = 10000 + export const TANSTACK_KEYS = { AUTH: ['auth'], USER: ['user'], diff --git a/frontend/src/pages/client/ClientTasksPage/components/ChatConditions/ChatConditions.tsx b/frontend/src/pages/client/ClientTasksPage/components/ChatConditions/ChatConditions.tsx index c1b4da39..c592bcf1 100644 --- a/frontend/src/pages/client/ClientTasksPage/components/ChatConditions/ChatConditions.tsx +++ b/frontend/src/pages/client/ClientTasksPage/components/ChatConditions/ChatConditions.tsx @@ -1,7 +1,7 @@ import { useState } from 'react' import { Block, List, useToast } from '@components' -import { useChat, useUserActions, useChatActions } from '@store' +import { TaskTimeoutError, useChat, useUserActions, useChatActions } from '@store' import { checkWalletRequirements } from '../../helpers' import { ChatConditionItem, WalletCondition } from './components' @@ -28,9 +28,25 @@ export const ChatConditions = () => { if (chat?.slug) { await fetchUserChatAction(chat.slug) } + showToast({ + type: 'success', + message: 'Gifts refreshed successfully', + }) } catch (e: unknown) { - const message = e instanceof Error ? e.message : 'Failed to refresh gifts' - showToast({ type: 'error', message }) + if (e instanceof TaskTimeoutError) { + showToast({ + type: 'warning', + message: + 'Your request to refresh gifts was submitted, but took longer than expected. Please, wait for 5 minutes or try again', + }) + if (chat?.slug) { + await fetchUserChatAction(chat.slug) + } + } else { + const message = + e instanceof Error ? e.message : 'Failed to refresh gifts' + showToast({ type: 'error', message }) + } } finally { setIsRefreshingGifts(false) } diff --git a/frontend/src/services/ApiService.ts b/frontend/src/services/ApiService.ts index 490df2ed..5de63b75 100644 --- a/frontend/src/services/ApiService.ts +++ b/frontend/src/services/ApiService.ts @@ -1,6 +1,6 @@ // src/services/ApiService.ts import { getValidationError } from '@utils' -import ky, { HTTPError, Options } from 'ky' +import ky, { HTTPError, Options, TimeoutError } from 'ky' import config from '@config' @@ -24,6 +24,7 @@ interface ApiOptions extends Options { statusCodes: number[] } cache?: 'force-cache' | 'no-cache' | 'no-store' | 'only-if-cached' | 'reload' + timeout?: number | false } type QueryParams = Record @@ -57,6 +58,14 @@ const handleError = async ( } } + if (err instanceof TimeoutError) { + return { + ok: false, + status: 408, + error: 'Request timed out', + } + } + if (err instanceof TypeError) { return { ok: false, @@ -107,9 +116,9 @@ export const ApiService = { host?: string }): Promise> => { try { - const { retry, cache, signal, ...searchParams } = options + const { retry, cache, signal, timeout, ...searchParams } = options const searchParamsValue = - Object.keys(options).length > 0 ? searchParams : undefined + Object.keys(searchParams).length > 0 ? searchParams : undefined const response = await api .get(`${host ?? apiHost}${endpoint}`, { searchParams: searchParamsValue as QueryParams, @@ -120,6 +129,7 @@ export const ApiService = { }, cache, signal, + timeout, }) .json() @@ -144,9 +154,9 @@ export const ApiService = { host?: string }): Promise> => { try { - const { retry, cache, signal, ...searchParams } = options + const { retry, cache, signal, timeout, ...searchParams } = options const searchParamsValue = - Object.keys(options).length > 0 ? searchParams : undefined + Object.keys(searchParams).length > 0 ? searchParams : undefined const response = await api .post(`${host ?? apiHost}${endpoint}`, { json: data, @@ -158,6 +168,7 @@ export const ApiService = { }, cache, signal, + timeout, }) .json() @@ -182,9 +193,9 @@ export const ApiService = { host?: string }): Promise> => { try { - const { retry, cache, signal, ...searchParams } = options + const { retry, cache, signal, timeout, ...searchParams } = options const searchParamsValue = - Object.keys(options).length > 0 ? searchParams : undefined + Object.keys(searchParams).length > 0 ? searchParams : undefined const response = await api .put(`${host ?? apiHost}${endpoint}`, { json: data, @@ -196,6 +207,7 @@ export const ApiService = { }, cache, signal, + timeout, }) .json() @@ -218,9 +230,9 @@ export const ApiService = { host?: string }): Promise> => { try { - const { retry, cache, signal, ...searchParams } = options + const { retry, cache, signal, timeout, ...searchParams } = options const searchParamsValue = - Object.keys(options).length > 0 ? searchParams : undefined + Object.keys(searchParams).length > 0 ? searchParams : undefined const response = await api .delete(`${host ?? apiHost}${endpoint}`, { searchParams: searchParamsValue as QueryParams, @@ -231,6 +243,7 @@ export const ApiService = { }, cache, signal, + timeout, }) .json() diff --git a/frontend/src/store/user/api.ts b/frontend/src/store/user/api.ts index cbce76d0..7cce9cfb 100644 --- a/frontend/src/store/user/api.ts +++ b/frontend/src/store/user/api.ts @@ -1,4 +1,5 @@ import { ApiService, ApiServiceResponse } from '@services' +import { ASYNC_TASK_TIMEOUT_MS } from '@utils' import { AuthenticatedUser, ChatTaskComplete, User, WalletData } from './types' @@ -68,6 +69,9 @@ export const completeChatTasksAPI = async ( ): Promise> => { const response = await ApiService.get({ endpoint: `/system/async-tasks/${taskId}`, + options: { + timeout: ASYNC_TASK_TIMEOUT_MS, + }, }) return response diff --git a/frontend/src/store/user/user.ts b/frontend/src/store/user/user.ts index 25ca8efd..97ee4ca1 100644 --- a/frontend/src/store/user/user.ts +++ b/frontend/src/store/user/user.ts @@ -36,6 +36,14 @@ interface UserActions { } } +export class TaskTimeoutError extends Error { + isTimeout = true + constructor(message = 'Task took longer than expected') { + super(message) + this.name = 'TaskTimeoutError' + } +} + const webApp = window.Telegram?.WebApp const useUserStore = create((set) => ({ @@ -91,9 +99,12 @@ const useUserStore = create((set) => ({ return data?.taskId }, completeChatTaskAction: async (taskId: string) => { - const { ok, error } = await completeChatTasksAPI(taskId) + const { ok, status, error } = await completeChatTasksAPI(taskId) if (!ok) { + if (status === 408) { + throw new TaskTimeoutError(error) + } throw new Error(error) } },