Skip to content

🎨 Increase function rpc call timeouts to 30sec #8440

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

Merged
merged 1 commit into from
Oct 1, 2025
Merged

🎨 Increase function rpc call timeouts to 30sec #8440

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
33 changes: 31 additions & 2 deletions ...ary/src/servicelib/rabbitmq/rpc_interfaces/webserver/functions/functions_rpc_interface.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import logging
from typing import Literal
from typing import Annotated, Literal

from models_library.api_schemas_webserver import WEBSERVER_RPC_NAMESPACE
from models_library.api_schemas_webserver.functions import (
Expand Down Expand Up @@ -32,13 +32,15 @@
from models_library.rest_ordering import OrderBy
from models_library.rest_pagination import PageMetaInfoLimitOffset
from models_library.users import UserID
from pydantic import TypeAdapter
from pydantic import PositiveInt, TypeAdapter

from .....logging_utils import log_decorator
from .... import RabbitMQRPCClient

_logger = logging.getLogger(__name__)

_FUNCTION_RPC_TIMEOUT_SEC: Annotated[int, PositiveInt] = 30


@log_decorator(_logger, level=logging.DEBUG)
async def register_function(
Expand All @@ -54,6 +56,7 @@ async def register_function(
function=function,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(RegisteredFunction).validate_python(
result
Expand All @@ -74,6 +77,7 @@ async def get_function(
function_id=function_id,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(RegisteredFunction).validate_python(result)

Expand All @@ -92,6 +96,7 @@ async def get_function_input_schema(
function_id=function_id,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(FunctionInputSchema).validate_python(result)

Expand All @@ -110,6 +115,7 @@ async def get_function_output_schema(
function_id=function_id,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(FunctionOutputSchema).validate_python(result)

Expand All @@ -128,6 +134,7 @@ async def delete_function(
function_id=function_id,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
assert result is None # nosec
return result
Expand Down Expand Up @@ -158,6 +165,7 @@ async def list_functions(
filter_by_function_class=filter_by_function_class,
search_by_function_title=search_by_function_title,
search_by_multi_columns=search_by_multi_columns,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
)
return TypeAdapter(
Expand Down Expand Up @@ -188,6 +196,7 @@ async def list_function_jobs(
filter_by_function_id=filter_by_function_id,
filter_by_function_job_ids=filter_by_function_job_ids,
filter_by_function_job_collection_id=filter_by_function_job_collection_id,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
)
return TypeAdapter(
Expand Down Expand Up @@ -220,6 +229,7 @@ async def list_function_jobs_with_status(
filter_by_function_id=filter_by_function_id,
filter_by_function_job_ids=filter_by_function_job_ids,
filter_by_function_job_collection_id=filter_by_function_job_collection_id,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(
tuple[
Expand Down Expand Up @@ -247,6 +257,7 @@ async def list_function_job_collections(
filters=filters,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(
tuple[list[RegisteredFunctionJobCollection], PageMetaInfoLimitOffset]
Expand All @@ -269,6 +280,7 @@ async def update_function_title(
title=title,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(RegisteredFunction).validate_python(result)

Expand All @@ -289,6 +301,7 @@ async def update_function_description(
description=description,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(RegisteredFunction).validate_python(result)

Expand All @@ -309,6 +322,7 @@ async def run_function(
inputs=inputs,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(RegisteredFunctionJob).validate_python(
result
Expand All @@ -329,6 +343,7 @@ async def register_function_job(
function_job=function_job,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(RegisteredFunctionJob).validate_python(
result
Expand All @@ -351,6 +366,7 @@ async def patch_registered_function_job(
product_name=product_name,
function_job_uuid=function_job_uuid,
registered_function_job_patch=registered_function_job_patch,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(RegisteredFunctionJob).validate_python(
result
Expand All @@ -371,6 +387,7 @@ async def get_function_job(
function_job_id=function_job_id,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)

return TypeAdapter(RegisteredFunctionJob).validate_python(result)
Expand All @@ -390,6 +407,7 @@ async def get_function_job_status(
function_job_id=function_job_id,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(FunctionJobStatus).validate_python(result)

Expand All @@ -408,6 +426,7 @@ async def get_function_job_outputs(
function_job_id=function_job_id,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(FunctionOutputs).validate_python(result)

Expand All @@ -430,6 +449,7 @@ async def update_function_job_status(
user_id=user_id,
product_name=product_name,
check_write_permissions=check_write_permissions,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(FunctionJobStatus).validate_python(result)

Expand All @@ -452,6 +472,7 @@ async def update_function_job_outputs(
user_id=user_id,
product_name=product_name,
check_write_permissions=check_write_permissions,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(FunctionOutputs).validate_python(result)

Expand All @@ -470,6 +491,7 @@ async def delete_function_job(
function_job_id=function_job_id,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
assert result is None # nosec

Expand All @@ -490,6 +512,7 @@ async def find_cached_function_jobs(
inputs=inputs,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
if result is None:
return None
Expand All @@ -510,6 +533,7 @@ async def register_function_job_collection(
function_job_collection=function_job_collection,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(RegisteredFunctionJobCollection).validate_python(result)

Expand All @@ -528,6 +552,7 @@ async def get_function_job_collection(
function_job_collection_id=function_job_collection_id,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(RegisteredFunctionJobCollection).validate_python(result)

Expand All @@ -546,6 +571,7 @@ async def delete_function_job_collection(
function_job_collection_id=function_job_collection_id,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
assert result is None # nosec

Expand All @@ -564,6 +590,7 @@ async def get_function_user_permissions(
function_id=function_id,
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(FunctionUserAccessRights).validate_python(result)

Expand All @@ -582,6 +609,7 @@ async def get_functions_user_api_access_rights(
),
user_id=user_id,
product_name=product_name,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(FunctionUserApiAccessRights).validate_python(result)

Expand Down Expand Up @@ -614,6 +642,7 @@ async def set_group_permissions(
read=read,
write=write,
execute=execute,
timeout_s=_FUNCTION_RPC_TIMEOUT_SEC,
)
return TypeAdapter(
list[tuple[FunctionID | FunctionJobID, FunctionGroupAccessRights]]
Expand Down
Loading