Release ownership of this channel safely
(self, channel_id)
| 489 | return False |
| 490 | |
| 491 | def release_ownership(self, channel_id): |
| 492 | """Release ownership of this channel safely""" |
| 493 | if not self.redis_client: |
| 494 | return |
| 495 | |
| 496 | try: |
| 497 | lock_key = RedisKeys.channel_owner(channel_id) |
| 498 | |
| 499 | # Only delete if we're the current owner to prevent race conditions |
| 500 | current = self.redis_client.get(lock_key) |
| 501 | if current and current == self.worker_id: |
| 502 | self.redis_client.delete(lock_key) |
| 503 | logger.info(f"Released ownership of channel {channel_id}") |
| 504 | |
| 505 | # Also ensure channel stopping key is set to signal clients |
| 506 | stop_key = RedisKeys.channel_stopping(channel_id) |
| 507 | self.redis_client.setex(stop_key, 30, "true") |
| 508 | logger.info(f"Set stopping signal for channel {channel_id} clients") |
| 509 | |
| 510 | except Exception as e: |
| 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""" |
no test coverage detected