diff --git a/controller/breeder_service.py b/controller/breeder_service.py index 8a798ed..f9d3df5 100755 --- a/controller/breeder_service.py +++ b/controller/breeder_service.py @@ -847,12 +847,25 @@ def delete_breeder(self, breeder_id, force=False): if failed_count > 0: logger.warning(f"{failed_count} worker jobs could not be cancelled") + # Clean coordination state before dropping the database + self.archive_repo.cleanup_coordination_state(breeder_id) + + # Read group before metadata is removed + det_cfg = breeder_config.get('interference_detection', breeder_config.get('detection', {})) + group_id = det_cfg.get('group', breeder_config.get('group', 'default')) + # Drop the archive database self.archive_repo.drop_database(__uuid_common_name) # Remove metadata self.metadata_repo.remove_breeder_meta(breeder_id) + # If this was the last breeder in the group, purge the lease row + remaining_in_group = self._count_breeders_in_group(group_id) + if remaining_in_group == 0: + self.archive_repo.cleanup_group_lease(group_id) + logger.info(f"Purged group lease — last breeder in group '{group_id}' deleted") + logger.info(f"Successfully deleted breeder: {breeder_id}") return { "result": "SUCCESS", @@ -908,6 +921,27 @@ def list_breeders(self): "error": str(e) } + def _count_breeders_in_group(self, group_id): + """Count remaining breeders in the same group after a deletion.""" + try: + self.metadata_repo.create_table() + breeders = self.metadata_repo.fetch_breeders_list() + count = 0 + for row in breeders: + bid = row[0] + meta = self.metadata_repo.fetch_meta_data(bid) + if meta and len(meta) > 0: + cfg = meta[0][3] + if isinstance(cfg, dict): + det_cfg = cfg.get('interference_detection', cfg.get('detection', {})) + bgroup = det_cfg.get('group', cfg.get('group', 'default')) + if bgroup == group_id: + count += 1 + return count + except Exception as e: + logger.warning(f"Failed to count breeders in group {group_id}: {e}") + return -1 + def _count_config_params(self, config): new_param_count = 0 for category in ['sysctl', 'sysfs', 'cpufreq', 'ethtool']: diff --git a/controller/database.py b/controller/database.py index 20ffa44..f7313d0 100644 --- a/controller/database.py +++ b/controller/database.py @@ -128,45 +128,43 @@ def drop_database(self, breeder_id): execute_ddl_query(db_config, query) logger.info(f"Dropped archive database: {breeder_id}") - def ensure_detection_rounds_table(self): - """Create the detection_rounds table in archive_db if it doesn't exist. - - The detection_rounds table coordinates impulse detection between breeders. - Each row represents a round where one breeder (sender) pushes an impulse - while all others hold still. The controller creates this table and inserts - rows at breeder creation time. + def cleanup_coordination_state(self, breeder_id): + """Remove a breeder's rows from coordination tables in archive_db. + + Called on breeder deletion. Removes the breeder from + interference_active_breeders and detection_readiness so stale + coordination state doesn't block other breeders in the group. """ db_config = self.base_config.copy() db_config['database'] = "archive_db" - query = """ - CREATE TABLE IF NOT EXISTS detection_rounds ( - round_id SERIAL PRIMARY KEY, - sender_id VARCHAR(255) NOT NULL, - status TEXT NOT NULL DEFAULT 'active', - created_at TIMESTAMPTZ DEFAULT NOW(), - completed_at TIMESTAMPTZ, - receiver_violated BOOLEAN NOT NULL DEFAULT FALSE - ); - CREATE INDEX IF NOT EXISTS idx_detection_rounds_active - ON detection_rounds (status) WHERE status = 'active'; - """ + queries = [ + f"DELETE FROM interference_active_breeders WHERE breeder_id = '{breeder_id}';", + f"DELETE FROM detection_readiness WHERE breeder_id = '{breeder_id}';", + ] - execute_query(db_config, query) - logger.info("Ensured detection_rounds table exists in archive_db") + for query in queries: + try: + execute_query(db_config, query) + except Exception as e: + logger.warning(f"Coordination cleanup query failed (table may not exist yet): {e}") - def insert_detection_round(self, sender_id): - """Insert a detection round for a sender breeder. - - Args: - sender_id: UUID of the breeder that will send the impulse + logger.info(f"Cleaned coordination state for breeder: {breeder_id}") + + def cleanup_group_lease(self, group_id): + """Remove the sender_lease row for a group with no remaining breeders. + + Called automatically when the last breeder in a group is deleted. + Per-breeder coordination rows (interference_active_breeders, + detection_readiness) are already cleaned by cleanup_coordination_state. """ db_config = self.base_config.copy() db_config['database'] = "archive_db" - query = f"INSERT INTO detection_rounds (sender_id) VALUES ('{sender_id}');" - execute_query(db_config, query) - logger.info(f"Inserted detection round for sender: {sender_id}") + try: + execute_query(db_config, f"DELETE FROM sender_lease WHERE group_id = '{group_id}';") + except Exception as e: + logger.warning(f"Group lease cleanup failed (table may not exist yet): {e}") def get_connection_url(self, breeder_id): """Get PostgreSQL connection URL for a breeder database""" diff --git a/tests/unit/test_detection_rounds.py b/tests/unit/test_detection_rounds.py deleted file mode 100644 index 1c77afa..0000000 --- a/tests/unit/test_detection_rounds.py +++ /dev/null @@ -1,107 +0,0 @@ -# -# Copyright (c) 2019 Matthias Tafelmeier. -# -# This file is part of godon -# -# godon is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as -# published by the Free Software Foundation, either version 3 of -# the License, or (at your option) any later version. -# -# godon is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this godon. If not, see . -# - -import pytest -import sys -import os -from unittest.mock import MagicMock, patch, call - -sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../..')) - -from controller.database import ArchiveDatabaseRepository, execute_query - - -class TestDetectionRoundsTable: - """Test detection_rounds table creation and row insertion""" - - def setup_method(self): - self.base_config = { - 'host': 'localhost', - 'port': '5432', - 'user': 'test_user', - 'password': 'test_pass', - 'database': 'archive_db', - } - self.repo = ArchiveDatabaseRepository(self.base_config) - - @patch('controller.database.execute_query') - def test_ensure_detection_rounds_table_creates_table(self, mock_exec): - """ensure_detection_rounds_table issues CREATE TABLE IF NOT EXISTS""" - self.repo.ensure_detection_rounds_table() - - assert mock_exec.call_count == 1 - call_args = mock_exec.call_args - - # Verify it targets archive_db - db_config = call_args[0][0] - assert db_config['database'] == 'archive_db' - - # Verify SQL contains table creation - sql = call_args[0][1] - assert 'CREATE TABLE IF NOT EXISTS detection_rounds' in sql - assert 'round_id' in sql - assert 'sender_id' in sql - assert 'status' in sql - assert 'created_at' in sql - assert 'completed_at' in sql - - @patch('controller.database.execute_query') - def test_ensure_detection_rounds_table_creates_index(self, mock_exec): - """ensure_detection_rounds_table creates active status index""" - self.repo.ensure_detection_rounds_table() - - sql = mock_exec.call_args[0][1] - assert 'idx_detection_rounds_active' in sql - assert "WHERE status = 'active'" in sql - - @patch('controller.database.execute_query') - def test_insert_detection_round_inserts_sender(self, mock_exec): - """insert_detection_round inserts a row with the sender UUID""" - sender_id = 'abc-123-def' - self.repo.insert_detection_round(sender_id) - - assert mock_exec.call_count == 1 - call_args = mock_exec.call_args - - db_config = call_args[0][0] - assert db_config['database'] == 'archive_db' - - sql = call_args[0][1] - assert 'INSERT INTO detection_rounds' in sql - assert 'sender_id' in sql - assert sender_id in sql - - @patch('controller.database.execute_query') - def test_insert_detection_round_uses_default_status(self, mock_exec): - """Inserted rows should rely on the DEFAULT 'active' status""" - self.repo.insert_detection_round('some-sender') - - sql = mock_exec.call_args[0][1] - assert "VALUES ('some-sender')" in sql - - @patch('controller.database.execute_query') - def test_ensure_table_idempotent(self, mock_exec): - """Calling ensure twice should not error (IF NOT EXISTS)""" - self.repo.ensure_detection_rounds_table() - self.repo.ensure_detection_rounds_table() - - assert mock_exec.call_count == 2 - # Both calls use IF NOT EXISTS — second is a no-op - sql = mock_exec.call_args[0][1] - assert 'IF NOT EXISTS' in sql