From e2157443b76fafe9a6c5007a665f82d80b92c6d6 Mon Sep 17 00:00:00 2001 From: eezd Date: Wed, 4 Nov 2026 16:59:08 +0800 Subject: [PATCH] Bound CLS cleanup time and support stopped-instance retries --- backend/src/alphapulse/crawler/remove_cls.py | 11 +++++++++ scripts/deploy.py | 24 +++++++++++--------- scripts/tests/test_remove_cls.py | 8 ++++++- 3 files changed, 31 insertions(+), 12 deletions(-) diff --git a/backend/src/alphapulse/crawler/remove_cls.py b/backend/src/alphapulse/crawler/remove_cls.py index 5d980c0..feb209f 100644 --- a/backend/src/alphapulse/crawler/remove_cls.py +++ b/backend/src/alphapulse/crawler/remove_cls.py @@ -64,6 +64,15 @@ async def cleanup(connection, execute=False): for table in SOURCE_TABLES}) counts['news_revisions'] = await connection.scalar(text('''SELECT count(*) FROM news_revisions WHERE news_id IN (SELECT id FROM news_items WHERE source_code = ANY(:sources))'''), params) + cleanup_indexes = { + 'cleanup_cls_revisions_raw': ('news_revisions', 'raw_response_id'), + 'cleanup_cls_economic_raw': ('economic_revisions', 'raw_response_id'), + 'cleanup_cls_series_raw': ('economic_series_values', 'raw_response_id'), + 'cleanup_cls_response_body': ('raw_responses', 'body_sha256'), + } + if execute: + for name, (table, column) in cleanup_indexes.items(): + await connection.execute(text(f'CREATE INDEX {name} ON {table} ({column})')) before = await retained_news(connection) protected = await protected_records(connection) if execute: @@ -83,6 +92,8 @@ async def cleanup(connection, execute=False): WHERE body.sha256 = removed.body_sha256 AND NOT EXISTS (SELECT 1 FROM raw_responses response WHERE response.body_sha256 = body.sha256)''')) counts['unreferenced_raw_bodies'] = result.rowcount + for name in cleanup_indexes: + await connection.execute(text('DROP INDEX ' + name)) if protected != await protected_records(connection): raise RuntimeError('保留业务数据校验失败，清理已回滚') if before != await retained_news(connection): diff --git a/scripts/deploy.py b/scripts/deploy.py index 6a74cdb..e60df91 100644 --- a/scripts/deploy.py +++ b/scripts/deploy.py @@ -270,18 +270,20 @@ def remove_cls(build, execute=False): if execute: current = json.loads((INSTANCE / 'deployment.json').read_text()) old_command = compose(INSTANCE, current) - for _ in range(60): + running = any(container['State']['Running'] for container in inspect_services(old_command)) + if running: + for _ in range(60): + idle = json.loads(run('docker', 'exec', '-i', 'AlphaPulse-dev-api', 'python', '-', input=IDLE)) + if not idle['busy']: + break + time.sleep(5) + else: + raise RuntimeError('Chat or trading analysis remained busy') + run('docker', 'pause', 'AlphaPulse-dev-trader') idle = json.loads(run('docker', 'exec', '-i', 'AlphaPulse-dev-api', 'python', '-', input=IDLE)) - if not idle['busy']: - break - time.sleep(5) - else: - raise RuntimeError('Chat or trading analysis remained busy') - run('docker', 'pause', 'AlphaPulse-dev-trader') - idle = json.loads(run('docker', 'exec', '-i', 'AlphaPulse-dev-api', 'python', '-', input=IDLE)) - if idle['busy']: - run('docker', 'unpause', 'AlphaPulse-dev-trader') - raise RuntimeError('Analysis became active; retry after it completes') + if idle['busy']: + run('docker', 'unpause', 'AlphaPulse-dev-trader') + raise RuntimeError('Analysis became active; retry after it completes') run(*old_command, 'stop') print(run(*command, 'python', '-m', 'alphapulse.crawler.remove_cls', *(['--execute'] if execute else [])), flush=True) if execute: diff --git a/scripts/tests/test_remove_cls.py b/scripts/tests/test_remove_cls.py index 2e7217d..692781a 100644 --- a/scripts/tests/test_remove_cls.py +++ b/scripts/tests/test_remove_cls.py @@ -9,7 +9,7 @@ import deploy class RemoveClsTests(unittest.TestCase): - def exercise(self, execute, failure=False): + def exercise(self, execute, failure=False, running=True): with tempfile.TemporaryDirectory() as temporary: directory = Path(temporary) build = {'commit': 'a'*40, 'version': '20261104-aaaaaaaa'} @@ -26,6 +26,7 @@ class RemoveClsTests(unittest.TestCase): with patch.object(deploy, 'INSTANCE', directory), patch.object(deploy, 'validate_images'), \ patch.object(deploy, 'branch_head', return_value=build['commit']), \ + patch.object(deploy, 'inspect_services', return_value=[{'State': {'Running': running}}]), \ patch.object(deploy, 'run', side_effect=command), patch.object(deploy, 'deploy') as update: if failure: with self.assertRaisesRegex(RuntimeError, 'cleanup failed'): @@ -54,3 +55,8 @@ class RemoveClsTests(unittest.TestCase): calls = self.exercise(True, True) self.assertTrue(any('stop' in args for args in calls)) self.assertFalse(any('up' in args or 'alembic' in args for args in calls)) + + def test_execute_resumes_with_all_writers_stopped(self): + calls = self.exercise(True, running=False) + self.assertFalse(any(args[:2] == ('docker', 'pause') for args in calls)) + self.assertTrue(any('--execute' in args for args in calls))