henry3260 commented on code in PR #65587:
URL: https://github.com/apache/airflow/pull/65587#discussion_r4012858829
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2014,26 +2015,38 @@ def in_process_api_server():
return api
+_IN_PROCESS_RESPONSE_SINK: ContextVar[deque[BaseModel | None] | None] =
ContextVar(
+ "in_process_response_sink", default=None
+)
+"""Where :meth:`InProcessTestSupervisor.send_msg` must deliver the response it
is about to send.
+
+Only :meth:`InProcessSupervisorComms.send` sets it, and it gets a fresh sink
per call, so a response
+can never reach a caller other than the one that is waiting for it. The socket
is read by the raw
+thread started in ``_setup_subprocess_socket``, which never has a sink set:
requests from a child
Review Comment:
> This says the socket thread never has a sink set, but every child
`GetVariable`/`GetConnection` handled on that thread calls `mask_secret`
(`request_handlers.py:106` and `:94`), which re-enters
`InProcessSupervisorComms.send` at `sdk/log.py:266` and sets a sink on that
very thread. The code survives it, since the nested token resets before the
outer `send_msg` runs and the nested `MaskSecret` reply lands in the sink
rather than on the child's socket, but this docstring is the only place the
design is written down and it is the sentence someone would rely on to replace
the sink with a thread-identity check. The class docstring just below at line
2032 also still says this handler "uses deques instead of sockets", which this
diff makes untrue in both halves.
Done!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
