Extend ownership lease, re-acquiring if key expired
(self, channel_id, ttl=30)
| 511 | logger.error(f"Error releasing channel ownership: {e}") |
| 512 | |
| 513 | def extend_ownership(self, channel_id, ttl=30): |
| 514 | """Extend ownership lease, re-acquiring if key expired""" |
| 515 | if not self.redis_client: |
| 516 | return False |
| 517 | |
| 518 | try: |
| 519 | lock_key = RedisKeys.channel_owner(channel_id) |
| 520 | current = self.redis_client.get(lock_key) |
| 521 | |
| 522 | if current is None: |
| 523 | # Key expired, re-acquire if we still run local upstream, but never |
| 524 | # during coordinated teardown (multi-worker reconnect-during-stop race). |
| 525 | if channel_id in self.stream_managers or channel_id in self._live_stream_managers: |
| 526 | if self._channel_unavailable_for_new_clients(channel_id): |
| 527 | logger.info( |
| 528 | f"Refusing ownership re-acquisition for {channel_id}; " |
| 529 | f"teardown or pending shutdown active" |
| 530 | ) |
| 531 | return False |
| 532 | |
| 533 | acquired = self.redis_client.set(lock_key, self.worker_id, nx=True, ex=ttl) |
| 534 | if acquired: |
| 535 | logger.warning(f"Re-acquired expired ownership for channel {channel_id}") |
| 536 | return True |
| 537 | else: |
| 538 | new_owner = self.redis_client.get(lock_key) |
| 539 | logger.warning(f"Could not re-acquire ownership for {channel_id}, new owner: {new_owner}") |
| 540 | return False |
| 541 | return False |
| 542 | |
| 543 | if current == self.worker_id: |
| 544 | self.redis_client.expire(lock_key, ttl) |
| 545 | return True |
| 546 | |
| 547 | return False |
| 548 | except Exception as e: |
| 549 | logger.error(f"Error extending ownership: {e}") |
| 550 | return False |
| 551 | |
| 552 | def initialize_channel(self, url, channel_id, user_agent=None, transcode=False, stream_id=None): |
| 553 | """Initialize a channel without redundant active key""" |
no test coverage detected