diff --git a/app/apps/tasks_impl/app_manage.py b/app/apps/tasks_impl/app_manage.py index e68dcac..d27134d 100644 --- a/app/apps/tasks_impl/app_manage.py +++ b/app/apps/tasks_impl/app_manage.py @@ -65,6 +65,10 @@ async def app_manage_task_impl( AppsServiceKeys.APP_MANAGE_CHANNEL_KEY, redis_url=redis_url ) action = "installing" if mode == InstallMode.ON else "uninstalling" + # Only this task's own lock may be released, and only its own run may + # broadcast FINISHED. Otherwise a duplicate request (lock already held) + # would delete the running install's lock and stop its listener early. + lock_acquired = False try: message = await notifier.connect() match message: @@ -119,6 +123,7 @@ async def app_manage_task_impl( ) return + lock_acquired = True logger.info(f"App {action} lock acquired for {action}ing app {id}") apps_impl = Apps() res = ( @@ -185,29 +190,35 @@ async def app_manage_task_impl( case Err(report): _log_notify_listeners_error(action, id, report) finally: - res = await release_update_lock( - key=AppsServiceKeys.APP_MANAGE_LOCK_KEY, redis=redis_client - ) - match res: - case Ok(_): - logger.info(f"App {action} lock released.") - case Err(message): - logger.error(f"Failed to release app {action} lock: {message.format()}") - res = await notifier.send_message( - key=AppsServiceKeys.APP_MANAGE_MESSAGE_KEY, - contents=AppManageTaskMessage( - id=id, - mode=mode, - state=AppManagementProcessState.FAILURE, - message=f"Failed to release app {action} lock:" - f" {message.frames[0].message}", - ).model_dump_json(), - ) - match res: - case Err(report): - _log_notify_listeners_error(action, id, report) + # Only release the lock and finish the run if this task actually + # acquired the lock; otherwise we'd interfere with the task that owns it. + if lock_acquired: + res = await release_update_lock( + key=AppsServiceKeys.APP_MANAGE_LOCK_KEY, redis=redis_client + ) + match res: + case Ok(_): + logger.info(f"App {action} lock released.") + case Err(message): + logger.error( + f"Failed to release app {action} lock: {message.format()}" + ) + res = await notifier.send_message( + key=AppsServiceKeys.APP_MANAGE_MESSAGE_KEY, + contents=AppManageTaskMessage( + id=id, + mode=mode, + state=AppManagementProcessState.FAILURE, + message=f"Failed to release app {action} lock:" + f" {message.frames[0].message}", + ).model_dump_json(), + ) + match res: + case Err(report): + _log_notify_listeners_error(action, id, report) + + await _send_finish_message(id, mode, notifier) - await _send_finish_message(id, mode, notifier) if redis_client: await redis_client.close() diff --git a/tests/test_app_manage_lock.py b/tests/test_app_manage_lock.py new file mode 100644 index 0000000..f1e5bad --- /dev/null +++ b/tests/test_app_manage_lock.py @@ -0,0 +1,104 @@ +""" +Regression test: a duplicate app-manage request must not touch the lock or +lifecycle of the install that is already running. + +app_manage_task_impl released APP_MANAGE_LOCK_KEY and broadcast a FINISHED +message from its `finally` block even on the early-return path where the lock +was already held by another task. That deleted the running install's lock +(allowing concurrent installs) and made its listener stop early. +""" + +import json + +from app.apps.models import AppId, InstallMode +from app.external.result_type.src.result.result import Ok +from app.apps.tasks_impl import app_manage + + +async def test_lock_already_held_does_not_release_or_finish(monkeypatch): + sent = [] + release_calls = [] + + class FakeNotifier: + def __init__(self, *args, **kwargs): + pass + + async def connect(self): + return Ok(None) + + async def send_message(self, key, contents): + sent.append(contents) + return Ok(None) + + async def fake_acquire(key, lock_ttl, redis): + return Ok(False) # lock already held by the running install + + async def fake_release(key, redis): + release_calls.append(key) + return Ok(None) + + monkeypatch.setattr(app_manage, "BaseChannelNotifier", FakeNotifier) + monkeypatch.setattr(app_manage, "acquire_lock", fake_acquire) + monkeypatch.setattr(app_manage, "release_update_lock", fake_release) + + await app_manage.app_manage_task_impl( + "redis://127.0.0.1:6379/0", AppId.MEMPOOL, InstallMode.ON, True + ) + + states = [json.loads(c)["state"] for c in sent] + + assert release_calls == [], ( + "must not release the lock it never acquired (belongs to the running install)" + ) + assert "finished" not in states, ( + "must not broadcast FINISHED and stop the running install's listener" + ) + # the client should still be told the task is already running + assert "failure" in states + + +async def test_acquired_lock_is_released_and_finished(monkeypatch): + """The task that actually holds the lock must always release it and send + FINISHED, otherwise the lock stays stuck until its TTL expires.""" + sent = [] + release_calls = [] + + class FakeNotifier: + def __init__(self, *args, **kwargs): + pass + + async def connect(self): + return Ok(None) + + async def send_message(self, key, contents): + sent.append(contents) + return Ok(None) + + async def fake_acquire(key, lock_ttl, redis): + return Ok(True) # lock acquired by this task + + async def fake_release(key, redis): + release_calls.append(key) + return Ok(None) + + async def empty_install(app_id): + # an async generator that yields nothing (install does nothing here) + return + yield # pragma: no cover + + class FakeApps: + def install_app(self, app_id): + return empty_install(app_id) + + monkeypatch.setattr(app_manage, "BaseChannelNotifier", FakeNotifier) + monkeypatch.setattr(app_manage, "acquire_lock", fake_acquire) + monkeypatch.setattr(app_manage, "release_update_lock", fake_release) + monkeypatch.setattr(app_manage, "Apps", FakeApps) + + await app_manage.app_manage_task_impl( + "redis://127.0.0.1:6379/0", AppId.MEMPOOL, InstallMode.ON, True + ) + + states = [json.loads(c)["state"] for c in sent] + assert release_calls == [app_manage.AppsServiceKeys.APP_MANAGE_LOCK_KEY] + assert "finished" in states