self.refresh_from_db(lock_for_update=True, session=session)
                self.state = TaskInstanceState.SUCCESS
            except TaskDeferred as defer:
                
                
                self._defer_task(defer=defer, session=session)
                self.log.info(
                    "Pausing task as DEFERRED. dag_id=%s, task_id=%s, execution_date=%s, start_date=%s",
                    self.dag_id,
                    self.task_id,
                    _date_or_empty(task_instance=self, attr="execution_date"),
                    _date_or_empty(task_instance=self, attr="start_date"),
                )
                if not test_mode:
                    session.add(Log(self.state, self))
                    session.merge(self)
                    session.commit()
                return TaskReturnCode.DEFERRED
            except AirflowSkipException as e:
                
                
                if e.args:
                    self.log.info(e)
                if not test_mode:
                    self.refresh_from_db(lock_for_update=True, session=session)
                self.state = TaskInstanceState.SKIPPED
            except AirflowRescheduleException as reschedule_exception:
                self._handle_reschedule(actual_start_date, reschedule_exception, test_mode, session=session)
                session.commit()
                return None