feat: improve gifts indexing queue priority, task timeouts, and refresh UX feedback

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:

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
@@ -44,10 +45,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 ---------------------

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()

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,
         }
     )

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:

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'],

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)
     }

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<string, string | number | boolean>
@@ -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<ApiServiceResponse<T>> => {
     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<T>()
 
@@ -144,9 +154,9 @@ export const ApiService = {
     host?: string
   }): Promise<ApiServiceResponse<T>> => {
     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<T>()
 
@@ -182,9 +193,9 @@ export const ApiService = {
     host?: string
   }): Promise<ApiServiceResponse<T>> => {
     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<T>()
 
@@ -218,9 +230,9 @@ export const ApiService = {
     host?: string
   }): Promise<ApiServiceResponse<T>> => {
     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<T>()
 

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<ApiServiceResponse<ChatTaskComplete>> => {
   const response = await ApiService.get<ChatTaskComplete>({
     endpoint: `/system/async-tasks/${taskId}`,
+    options: {
+      timeout: ASYNC_TASK_TIMEOUT_MS,
+    },
   })
 
   return response

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<UserStore & UserActions>((set) => ({
@@ -91,9 +99,12 @@ const useUserStore = create<UserStore & UserActions>((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)
       }
     },