Repository: apache/airflow
File: airflow-core/src/airflow/jobs/triggerer_job_runner.py (Python)

Code before:
                    # TODO: better formatting of the exception?
                    self.failed_triggers.append((trigger_id, saved_exc))
                del self.triggers[trigger_id]
            await asyncio.sleep(0)
        return finished_ids

    def validate_state_changes(self, finished_ids: list[int]) -> messages.TriggerStateChanges:
        # Copy out of our dequeues in threadsafe manner to sync state with parent

        req_encoder = _new_encoder()
        events_to_send: list[tuple[int, DiscrimatedTriggerEvent]] = []
        failures_to_send: list[tuple[int, list[str] | None]] = []

        while self.events:
            trigger_id, trigger_event = self.events.popleft()

            try:
                req_encoder.encode(trigger_event)
                events_to_send.append((trigger_id, trigger_event))
            except NotImplementedError as e:
                logger.error(
                    "Trigger %s returned non-serializable result %r. Cancelling trigger.",
                    trigger_id,
                    trigger_event,
                )
                self.failed_triggers.append((trigger_id, e))

        while self.failed_triggers:
            trigger_id, exc = self.failed_triggers.popleft()
            tb = format_exception(type(exc), exc, exc.__traceback__) if exc else None
            failures_to_send.append((trigger_id, tb))

        return messages.TriggerStateChanges(
            events=events_to_send if events_to_send else None,
            finished=finished_ids if finished_ids else None,
            failures=failures_to_send if failures_to_send else None,
        )

    async def sync_state_to_supervisor(self, finished_ids: list[int]):
        msg = self.validate_state_changes(finished_ids=finished_ids)

        # Tell the monitor that we've finished triggers so it can update things
        resp = await self.send_state_changes(msg)

        if resp:
            self.to_create.extend(resp.to_create)
            self.to_cancel.extend(resp.to_cancel)

    async def send_state_changes(self, msg: messages.TriggerStateChanges) -> messages.TriggerStateSync | None:
        try:
            response = await self.comms_decoder.asend(msg)

Review comment (bug):
Probably best to catch any error.

Though I don't love the fact that this encodes things twice, throwing the result away once.

```suggestion
            try:
                req_encoder.encode(trigger_event)
            except Exception as e:
                logger.error(
                    "Trigger %s returned non-serializable result %r. Cancelling trigger.",
                    trigger_id,
                    trigger_event,
                )
                self.failed_triggers.append((trigger_id, e))
            else:
                events_to_send.append((trigger_id, trigger_event))

```

Code after:
                # any task instances depending on it need to be failed
                if details["events"] == 0:
                    self.log.error(
                        "Trigger exited without sending an event. Dependent tasks will be failed.",
                        name=details["name"],
                        trigger_id=trigger_id,
                    )
                    # TODO: better formatting of the exception?
                    self.failed_triggers.append((trigger_id, saved_exc))
                del self.triggers[trigger_id]
            await asyncio.sleep(0)
        return finished_ids

    def process_trigger_events(self, finished_ids: list[int]) -> messages.TriggerStateChanges:
        # Copy out of our dequeues in threadsafe manner to sync state with parent
        events_to_send: list[tuple[int, DiscrimatedTriggerEvent]] = []
        failures_to_send: list[tuple[int, list[str] | None]] = []

        while self.events:
            trigger_id, trigger_event = self.events.popleft()
            events_to_send.append((trigger_id, trigger_event))

        while self.failed_triggers:
            trigger_id, exc = self.failed_triggers.popleft()
            tb = format_exception(type(exc), exc, exc.__traceback__) if exc else None
            failures_to_send.append((trigger_id, tb))

        return messages.TriggerStateChanges(
            events=events_to_send if events_to_send else None,
            finished=finished_ids if finished_ids else None,
            failures=failures_to_send if failures_to_send else None,
        )

    def sanitize_trigger_events(self, msg: messages.TriggerStateChanges) -> messages.TriggerStateChanges:
        req_encoder = _new_encoder()
        events_to_send: list[tuple[int, DiscrimatedTriggerEvent]] = []

        if msg.events:
            for trigger_id, trigger_event in msg.events:
                try:
                    req_encoder.encode(trigger_event)
                except Exception as e:
                    logger.error(
                        "Trigger %s returned non-serializable result %r. Cancelling trigger.",
                        trigger_id,
                        trigger_event,
                    )
                    self.failed_triggers.append((trigger_id, e))
                else:
                    events_to_send.append((trigger_id, trigger_event))