From 33177bdb3c47ac5dbcf65c6de207b75eb0c224e2 Mon Sep 17 00:00:00 2001 From: Akanksha Trehun Date: Sat, 11 Jul 2026 20:32:38 +0530 Subject: [PATCH] Fix timeout logic bug and use time.monotonic() in openai batch trigger The timeout error message computed time.time() - self.end_time which produces a negative number since end_time is already in the past. Replace with a clear message stating the batch did not complete within the configured timeout. Also switch to time.monotonic() for the timeout check to align with the project coding standard. Signed-off-by: Akanksha Trehun --- .../openai/src/airflow/providers/openai/operators/openai.py | 2 +- .../openai/src/airflow/providers/openai/triggers/openai.py | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/providers/openai/src/airflow/providers/openai/operators/openai.py b/providers/openai/src/airflow/providers/openai/operators/openai.py index 0f040a03b5505..103f4c8ec60ee 100644 --- a/providers/openai/src/airflow/providers/openai/operators/openai.py +++ b/providers/openai/src/airflow/providers/openai/operators/openai.py @@ -194,7 +194,7 @@ def execute(self, context: Context) -> str | None: conn_id=self.conn_id, batch_id=self.batch_id, poll_interval=60, - end_time=time.time() + self.timeout, + end_time=time.monotonic() + self.timeout, ), method_name="execute_complete", ) diff --git a/providers/openai/src/airflow/providers/openai/triggers/openai.py b/providers/openai/src/airflow/providers/openai/triggers/openai.py index 5800e948490d3..05d6d5e14a441 100644 --- a/providers/openai/src/airflow/providers/openai/triggers/openai.py +++ b/providers/openai/src/airflow/providers/openai/triggers/openai.py @@ -58,12 +58,12 @@ async def run(self) -> AsyncIterator[TriggerEvent]: hook = OpenAIHook(conn_id=self.conn_id) try: while (batch := hook.get_batch(self.batch_id)) and BatchStatus.is_in_progress(batch.status): - if self.end_time < time.time(): + if self.end_time < time.monotonic(): yield TriggerEvent( { "status": "error", - "message": f"Batch {self.batch_id} has not reached a terminal status after " - f"{time.time() - self.end_time} seconds.", + "message": f"Batch {self.batch_id} has not reached a terminal status within " + f"the configured timeout.", "batch_id": self.batch_id, } )