Repository: apache/airflow
File: task-sdk/src/airflow/sdk/bases/sensor.py (Python)

Code before:
                    xcom_value = poke_return.xcom_value
                break

            if run_duration() > self.timeout:
                # If sensor is in soft fail mode but times out raise AirflowSkipException.
                message = (
                    f"Sensor has timed out; run duration of {run_duration()} seconds exceeds "
                    f"the specified timeout of {self.timeout}."
                )

                if self.soft_fail:
                    raise AirflowSkipException(message)
                raise AirflowSensorTimeout(message)
            if self.reschedule:
                next_poke_interval = self._get_next_poke_interval(started_at, run_duration, poke_count)
                reschedule_date = timezone.utcnow() + timedelta(seconds=next_poke_interval)
                raise AirflowRescheduleException(reschedule_date)
            time.sleep(self._get_next_poke_interval(started_at, run_duration, poke_count))
            poke_count += 1
        self.log.info("Success criteria met. Exiting.")
        return xcom_value

    def resume_execution(self, next_method: str, next_kwargs: dict[str, Any] | None, context: Context):
        try:
            try:
                return super().resume_execution(next_method, next_kwargs, context)
            except TaskDeferralTimeout as e:
                raise AirflowSensorTimeout(*e.args) from e
        except (AirflowException, TaskDeferralError) as e:
            if self.soft_fail:
                raise AirflowSkipException("Skipping due to soft_fail is set to True.") from e
            if self.never_fail:
                raise AirflowSkipException("Skipping due to never_fail is set to True.") from e
            raise

    def _get_next_poke_interval(
        self,
        started_at: datetime.datetime | float,
        run_duration: Callable[[], float],
        poke_count: int,
    ) -> float:
        """Use similar logic which is used for exponential backoff retry delay for operators."""
        if not self.exponential_backoff:
            return self.poke_interval

        if self.reschedule:
            # Calculate elapsed time since the sensor started
            elapsed_time = run_duration()

            # Initialize variables for the simulation
            cumulative_time: float = 0.0

Review comment (bug):
I would add a comment here explaining why there are nested try/except blocks. Perhaps something like:

`Nested try/except ensures TaskDeferralTimeout conversion does not bypass soft_fail / never_fail handling.`

Makes it easier to reason about. 

Code after:
                    xcom_value = poke_return.xcom_value
                break

            if run_duration() > self.timeout:
                # If sensor is in soft fail mode but times out raise AirflowSkipException.
                message = (
                    f"Sensor has timed out; run duration of {run_duration()} seconds exceeds "
                    f"the specified timeout of {self.timeout}."
                )

                if self.soft_fail:
                    raise AirflowSkipException(message)
                raise AirflowSensorTimeout(message)
            if self.reschedule:
                next_poke_interval = self._get_next_poke_interval(started_at, run_duration, poke_count)
                reschedule_date = timezone.utcnow() + timedelta(seconds=next_poke_interval)
                raise AirflowRescheduleException(reschedule_date)
            time.sleep(self._get_next_poke_interval(started_at, run_duration, poke_count))
            poke_count += 1
        self.log.info("Success criteria met. Exiting.")
        return xcom_value

    def resume_execution(self, next_method: str, next_kwargs: dict[str, Any] | None, context: Context):
        # Use nested try/except to convert TaskDeferralTimeout to AirflowSensorTimeout
        # while still allowing soft_fail/never_fail to handle both exception types.
        try:
            try:
                return super().resume_execution(next_method, next_kwargs, context)
            except TaskDeferralTimeout as e:
                raise AirflowSensorTimeout(*e.args) from e
        except (AirflowException, TaskDeferralError) as e:
            if self.soft_fail:
                raise AirflowSkipException("Skipping due to soft_fail is set to True.") from e
            if self.never_fail:
                raise AirflowSkipException("Skipping due to never_fail is set to True.") from e
            raise

    def _get_next_poke_interval(
        self,
        started_at: datetime.datetime | float,
        run_duration: Callable[[], float],
        poke_count: int,
    ) -> float:
        """Use similar logic which is used for exponential backoff retry delay for operators."""
        if not self.exponential_backoff:
            return self.poke_interval

        if self.reschedule:
            # Calculate elapsed time since the sensor started
            elapsed_time = run_duration()