Stop a channel with proper ownership handling
(self, channel_id)
| 1640 | logger.error(f"Error during forced cleanup for channel {channel_id}: {e}") |
| 1641 | |
| 1642 | def stop_channel(self, channel_id): |
| 1643 | """Stop a channel with proper ownership handling""" |
| 1644 | if channel_id in self._stopping_channels: |
| 1645 | logger.debug(f"stop_channel already in progress for {channel_id}, ignoring duplicate call") |
| 1646 | return |
| 1647 | self._stopping_channels.add(channel_id) |
| 1648 | self._stopping_since[channel_id] = time.time() |
| 1649 | stop_event_data = None |
| 1650 | redis_cleaned = False |
| 1651 | try: |
| 1652 | logger.info(f"Stopping channel {channel_id}") |
| 1653 | |
| 1654 | if self.redis_client: |
| 1655 | stop_key = RedisKeys.channel_stopping(channel_id) |
| 1656 | if self.redis_client.exists(stop_key): |
| 1657 | self.redis_client.expire(stop_key, 60) |
| 1658 | else: |
| 1659 | self.redis_client.setex(stop_key, 60, "true") |
| 1660 | |
| 1661 | was_owner = self.am_i_owner(channel_id) |
| 1662 | if was_owner: |
| 1663 | logger.info( |
| 1664 | f"This worker ({self.worker_id}) is the owner - closing provider connection" |
| 1665 | ) |
| 1666 | stop_event_data = self._collect_channel_stop_event_data(channel_id) |
| 1667 | |
| 1668 | # Stop new chunk writes before Redis cleanup; do not block on ffmpeg join yet. |
| 1669 | self._signal_upstream_shutdown(channel_id) |
| 1670 | |
| 1671 | # Release profile slots and delete Redis keys before any blocking local stop. |
| 1672 | # A concurrent disconnect + cleanup-thread stop used to wedge here behind a |
| 1673 | # 2s stderr join, never reaching scan/delete (bare buffer:index with TTL -1). |
| 1674 | self._clean_redis_keys(channel_id) |
| 1675 | redis_cleaned = True |
| 1676 | |
| 1677 | if was_owner: |
| 1678 | self.release_ownership(channel_id) |
| 1679 | |
| 1680 | self._stop_local_stream_activity(channel_id) |
| 1681 | self._spawn_channel_stop_event(stop_event_data) |
| 1682 | |
| 1683 | return True |
| 1684 | except Exception as e: |
| 1685 | logger.error(f"Error stopping channel {channel_id}: {e}") |
| 1686 | return False |
| 1687 | finally: |
| 1688 | if not redis_cleaned: |
| 1689 | try: |
| 1690 | self._clean_redis_keys(channel_id) |
| 1691 | except Exception as e: |
| 1692 | logger.error( |
| 1693 | f"Error cleaning Redis keys for channel {channel_id} during finally: {e}" |
| 1694 | ) |
| 1695 | self._stopping_channels.discard(channel_id) |
| 1696 | self._stopping_since.pop(channel_id, None) |
| 1697 | |
| 1698 | def check_inactive_channels(self): |
| 1699 | """Check for inactive channels (no clients) and stop them""" |
no test coverage detected