2020 function_id: str,
2121 inputs: list[tuple[str, Datum]], # (param name, decoded Datum)
2222 trigger_metadata: dict[str, Datum],
23- ) -> tuple[bool, Optional[Datum], list[tuple[str, Datum]], Optional[str]]
24- # (ok, return_datum, output_data, exception_text)
23+ ) -> tuple[bool, Optional[Datum], list[tuple[str, Datum]],
24+ Optional[str], Optional[str]]
25+ # (ok, return_datum, output_data, exception_text, exception_traceback)
2526
2627Scope: this native path supports the classic (non http-v2) binding path,
2728sync and async functions, input/output bindings and $return. Deferred (SDK-type)
2829bindings and the http-v2 streaming path are intentionally out of scope here and
2930fall back through the protobuf control path if ever needed.
3031"""
3132import sys
33+ import traceback
3234
3335from .bindings .context import get_context
3436from .bindings .meta import get_binding , get_datum , is_trigger_binding
@@ -151,7 +153,7 @@ async def invocation_request_native(invocation_id, function_id, inputs,
151153 threadpool = get_threadpool_executor ()
152154 fi = _functions .get_function (function_id )
153155 if fi is None :
154- return (False , None , [], "function %s not loaded" % function_id )
156+ return (False , None , [], "function %s not loaded" % function_id , "" )
155157
156158 try :
157159 args = _decode_inputs (fi , inputs , trigger_metadata )
@@ -188,32 +190,34 @@ async def invocation_request_native(invocation_id, function_id, inputs,
188190 return_datum = _encode_return (fi , call_result )
189191
190192 sys .stdout .flush ()
191- return (True , return_datum , output_data , None )
193+ return (True , return_datum , output_data , None , None )
192194
193195 except Exception as ex : # noqa - surfaced to the Host as a failed invocation
194196 logger .exception ("Native invocation failed" )
195- return (False , None , [], repr (ex ))
197+ return (False , None , [], repr (ex ), traceback . format_exc () )
196198
197199
198200def run_invocation_sync (invocation_id , function_id , inputs , trigger_metadata ,
199201 trace_parent = "" , trace_state = "" , retry_count = 0 ,
200202 max_retry_count = 0 , retry_exception = None ):
201203 """Synchronous native invocation for *sync* customer functions.
202204
203- Runs the handler directly on the calling thread (no asyncio loop, no
204- ``run_in_executor`` bounce). With the Rust transport dispatching each call
205- on its own thread, this avoids funneling sync handlers through a single
206- event loop, trimming per-invocation overhead on the hot path .
205+ Queues the handler on the runtime's configured threadpool without
206+ funneling the invocation through the asyncio loop. This preserves
207+ ``PYTHON_THREADPOOL_THREAD_COUNT`` for the native path while keeping datum
208+ conversion on the Rust-dispatched calling thread .
207209
208- Returns ``(handled, ok, return_datum, output_data, exception_text)``.
210+ Returns ``(handled, ok, return_datum, output_data, exception_text,
211+ exception_traceback)``.
209212 ``handled=False`` means the function is ``async`` and the caller should use
210213 the coroutine path (:func:`invocation_request_native`) instead.
211214 """
212215 fi = _functions .get_function (function_id )
213216 if fi is None :
214- return (True , False , None , [], "function %s not loaded" % function_id )
217+ return (True , False , None , [],
218+ "function %s not loaded" % function_id , "" )
215219 if fi .is_async :
216- return (False , None , None , [], None )
220+ return (False , None , None , [], None , None )
217221
218222 try :
219223 args = _decode_inputs (fi , inputs , trigger_metadata )
@@ -223,14 +227,18 @@ def run_invocation_sync(invocation_id, function_id, inputs, trigger_metadata,
223227
224228 # run_sync_func calls configure_opentelemetry(fi_context) when otel is
225229 # available, so the trace context prepared above parents customer spans.
226- call_result = run_sync_func (invocation_id , fi_context , fi .func , args )
230+ threadpool = get_threadpool_executor ()
231+ if threadpool is None :
232+ raise RuntimeError ("threadpool executor is not initialized" )
233+ call_result = threadpool .submit (
234+ run_sync_func , invocation_id , fi_context , fi .func , args ).result ()
227235
228236 _check_return (fi , call_result )
229237 output_data = _collect_output (fi , args )
230238 return_datum = _encode_return (fi , call_result )
231239
232- return (True , True , return_datum , output_data , None )
240+ return (True , True , return_datum , output_data , None , None )
233241
234242 except Exception as ex : # noqa - surfaced to the Host as a failed invocation
235243 logger .exception ("Native sync invocation failed" )
236- return (True , False , None , [], repr (ex ))
244+ return (True , False , None , [], repr (ex ), traceback . format_exc () )
0 commit comments