👌 Allow config downgrade for known migrations

Allow `verdi config downgrade` to run when the current AiiDA
version cannot load the configuration for normal use, but still has
the migration path needed to downgrade it.

Introduce a maximum downgradeable configuration version that is
separate from the normally supported version, and use it to detect
when the CLI can safely skip loading the default profile.

The idea is that we can patch older versions to still include the
migration.

Author: agoscinski

src/aiida/cmdline/utils/defaults.py

@@ -25,6 +25,10 @@ def get_default_profile():
     """
     try:
         config = get_config(create=True)
+    except exceptions.ConfigurationVersionError as exception:
+        if exception.can_downgrade:
+            return None
+        echo.echo_critical(str(exception))
     except exceptions.ConfigurationError as exception:
         echo.echo_critical(str(exception))
 

src/aiida/common/exceptions.py

@@ -181,9 +181,16 @@ class MissingConfigurationError(ConfigurationError):
 
 
 class ConfigurationVersionError(ConfigurationError):
-    """Configuration error raised when the configuration file version is not
-    compatible with the current version.
-    """
+    """Configuration error raised when the configuration file version is not compatible with the current version."""
+
+    def __init__(self, message: str, *, can_downgrade: bool = False):
+        """Construct the exception.
+
+        :param message: The exception message.
+        :param can_downgrade: Whether this AiiDA version can downgrade the configuration to its supported version.
+        """
+        super().__init__(message)
+        self.can_downgrade = can_downgrade
 
 
 class ClosedStorage(AiidaException):

src/aiida/manage/configuration/migrations/migrations.py

@@ -35,6 +35,8 @@
 
 CURRENT_CONFIG_VERSION = 10
 OLDEST_COMPATIBLE_CONFIG_VERSION = 10
+# Highest configuration version for which this code can run downgrade migrations, even if it cannot load it.
+MAXIMUM_DOWNGRADE_CONFIG_VERSION = 10
 
 CONFIG_LOGGER = AIIDA_LOGGER.getChild('config')
 
@@ -498,6 +500,41 @@ def get_oldest_compatible_version(config):
     return config.get('CONFIG_VERSION', {}).get('OLDEST_COMPATIBLE', 0)
 
 
+def config_can_be_downgraded(
+    config: ConfigType,
+    target: int | None = None,
+    migrations: Iterable[type[SingleMigration]] = MIGRATIONS,
+) -> bool:
+    """Return whether the configuration can be downgraded to the target version.
+
+    This is intentionally distinct from :data:`CURRENT_CONFIG_VERSION`: an AiiDA version may not be able to load a
+    configuration for normal operation, but may still know enough migrations to rewrite it for an older version.
+
+    :param config: the configuration dictionary
+    :param target: the version to downgrade to, defaulting to :data:`CURRENT_CONFIG_VERSION`
+    :param migrations: the registered migrations to consider
+    :return: ``True`` if a chain of migrations exists to downgrade the configuration to the target version
+    """
+    target = CURRENT_CONFIG_VERSION if target is None else target
+    current = get_current_version(config)
+
+    if current <= target or current > MAXIMUM_DOWNGRADE_CONFIG_VERSION:
+        return False
+
+    used = []
+    while current > target:
+        try:
+            migrator = next(m for m in migrations if m.up_revision == current)
+        except StopIteration:
+            return False
+        if migrator in used:
+            return False
+        used.append(migrator)
+        current = migrator.down_revision
+
+    return current == target
+
+
 def upgrade_config(
     config: ConfigType, target: int = CURRENT_CONFIG_VERSION, migrations: Iterable[type[SingleMigration]] = MIGRATIONS
 ) -> ConfigType:
@@ -535,6 +572,13 @@ def downgrade_config(
     :return: the migrated configuration dictionary
     """
     current = get_current_version(config)
+    if current > MAXIMUM_DOWNGRADE_CONFIG_VERSION:
+        msg = (
+            f'Cannot downgrade configuration version {current}: this AiiDA version can only downgrade configuration '
+            f'versions up to {MAXIMUM_DOWNGRADE_CONFIG_VERSION}.'
+        )
+        raise exceptions.ConfigurationError(msg)
+
     used = []
     while current > target:
         current = get_current_version(config)
@@ -581,16 +625,23 @@ def config_needs_migrating(config, filepath: str | None = None):
 
     if oldest_compatible_version > CURRENT_CONFIG_VERSION:
         filepath = filepath if filepath else ''
-        msg = (
-            f'The AiiDA configuration file {filepath} has version {current_version}, which is not compatible with '
-            f'the current AiiDA version that supports configuration versions up to {CURRENT_CONFIG_VERSION}. '
-            'Before switching to an older AiiDA version, stop the daemon and avoid other AiiDA interactions. '
-            'Then use a newer AiiDA version that still supports '
-            f'configuration version {current_version} and run `verdi config downgrade {CURRENT_CONFIG_VERSION}` '
-            f'to rewrite the configuration file to version {CURRENT_CONFIG_VERSION}. See '
-            f'{URL_CONFIG_SCHEMA_COMPATIBILITY} for the compatibility '
-            'table.'
-        )
-        raise exceptions.ConfigurationVersionError(msg)
+        can_downgrade = config_can_be_downgraded(config)
+        if can_downgrade:
+            msg = (
+                f'The AiiDA configuration file {filepath} has version {current_version} which is not compatible with '
+                f'the current aiida version supporting up to version {CURRENT_CONFIG_VERSION}. '
+                f'Run `verdi config downgrade {CURRENT_CONFIG_VERSION}` to rewrite it for this version. See '
+                f'{URL_CONFIG_SCHEMA_COMPATIBILITY} for the compatibility table.'
+            )
+        else:
+            msg = (
+                f'The AiiDA configuration file {filepath} has version {current_version} which is not compatible with '
+                f'the current aiida version supporting up to version {CURRENT_CONFIG_VERSION}. '
+                'Before switching to an older AiiDA version, use a newer AiiDA version that supports '
+                f'configuration version {current_version} and run `verdi config downgrade {CURRENT_CONFIG_VERSION}` '
+                'to rewrite it for this version. See '
+                f'{URL_CONFIG_SCHEMA_COMPATIBILITY} for the compatibility table.'
+            )
+        raise exceptions.ConfigurationVersionError(msg, can_downgrade=can_downgrade)
 
     return CURRENT_CONFIG_VERSION > current_version

tests/cmdline/commands/test_config.py

@@ -8,6 +8,9 @@
 ###########################################################################
 """Tests for ``verdi config``."""
 
+import json
+import pathlib
+
 import pytest
 
 from aiida import get_profile
@@ -274,15 +277,27 @@ def test_config_downgrade(run_cli_command, config_with_profile_factory):
     assert 'Success: Downgraded' in result.output.strip()
 
 
-@pytest.mark.presto
-def test_config_list_advanced_flag(run_cli_command, config_with_profile_factory):
-    """Test that ``verdi config list --advanced`` flag is accepted and footer hint is shown."""
-    config_with_profile_factory()
+def test_config_downgrade_incompatible_but_downgradeable(run_cli_command, config_with_profile_factory, monkeypatch):
+    """Test `verdi config downgrade` can run even if the config cannot be loaded normally."""
+    from aiida.manage import configuration
+    from aiida.manage.configuration.migrations import migrations
+
+    config = config_with_profile_factory()
+    filepath = pathlib.Path(config.filepath)
+    dictionary = config.dictionary
+    dictionary['CONFIG_VERSION'] = {'CURRENT': 10, 'OLDEST_COMPATIBLE': 10}
+    dictionary.setdefault('options', {})['broker.task_timeout'] = 12
+    filepath.write_text(json.dumps(dictionary), encoding='utf8')
+
+    configuration.CONFIG = None
+    monkeypatch.setattr(migrations, 'CURRENT_CONFIG_VERSION', 9)
+    monkeypatch.setattr(migrations, 'MAXIMUM_DOWNGRADE_CONFIG_VERSION', 10)
 
-    result_default = run_cli_command(cmd_verdi.verdi, ['config', 'list'], use_subprocess=False)
-    assert '--advanced' in result_default.output
-    assert 'logging.sqlalchemy_loglevel' not in result_default.output
+    result = run_cli_command(
+        cmd_verdi.verdi, ['config', 'downgrade', '9'], initialize_ctx_obj=False, use_subprocess=False
+    )
 
-    result_advanced = run_cli_command(cmd_verdi.verdi, ['config', 'list', '--advanced'], use_subprocess=False)
-    assert 'Use `verdi config list --advanced`' not in result_advanced.output
-    assert 'logging.db_loglevel' not in result_advanced.output
+    assert 'Success: Downgraded' in result.output.strip()
+    dictionary = json.loads(filepath.read_text(encoding='utf8'))
+    assert dictionary['CONFIG_VERSION'] == {'CURRENT': 9, 'OLDEST_COMPATIBLE': 9}
+    assert dictionary['options']['rmq.task_timeout'] == 12

tests/manage/configuration/migrations/test_migrations.py

@@ -18,9 +18,10 @@
 from aiida.common.exceptions import ConfigurationError, ConfigurationVersionError
 from aiida.manage.configuration.migrations import check_and_migrate_config
 from aiida.manage.configuration.migrations.migrations import (
-    CURRENT_CONFIG_VERSION,
+    MAXIMUM_DOWNGRADE_CONFIG_VERSION,
     MIGRATIONS,
     Initial,
+    config_can_be_downgraded,
     downgrade_config,
     upgrade_config,
 )
@@ -70,15 +71,36 @@ def test_downgrade_path_fail(load_config_sample):
         downgrade_config(config_initial, 4, migrations=[CircularMigration])
 
 
-def test_config_needs_migrating_incompatible_version():
+def test_config_needs_migrating_incompatible_version(monkeypatch):
     """An incompatible configuration version should raise with downgrade guidance."""
+    from aiida.manage.configuration.migrations import migrations
+
+    # Lower the loadable version below the downgrade cap so a config at the cap is unloadable but downgradeable,
+    # exercising the can_downgrade guidance path (config_needs_migrating downgrades to CURRENT_CONFIG_VERSION).
+    monkeypatch.setattr(migrations, 'CURRENT_CONFIG_VERSION', MAXIMUM_DOWNGRADE_CONFIG_VERSION - 1)
     config = {
-        'CONFIG_VERSION': {'CURRENT': CURRENT_CONFIG_VERSION + 1, 'OLDEST_COMPATIBLE': CURRENT_CONFIG_VERSION + 1}
+        'CONFIG_VERSION': {
+            'CURRENT': MAXIMUM_DOWNGRADE_CONFIG_VERSION,
+            'OLDEST_COMPATIBLE': MAXIMUM_DOWNGRADE_CONFIG_VERSION,
+        }
     }
 
-    with pytest.raises(ConfigurationVersionError, match=r'verdi config downgrade'):
+    with pytest.raises(ConfigurationVersionError, match=r'verdi config downgrade') as exception:
         check_and_migrate_config(config, filepath='/tmp/config.json')
 
+    assert exception.value.can_downgrade
+
+
+def test_config_can_be_downgraded():
+    """Test detecting whether a configuration can be downgraded by this AiiDA version."""
+    assert config_can_be_downgraded(
+        {'CONFIG_VERSION': {'CURRENT': MAXIMUM_DOWNGRADE_CONFIG_VERSION, 'OLDEST_COMPATIBLE': 0}},
+        target=MAXIMUM_DOWNGRADE_CONFIG_VERSION - 1,
+    )
+    assert not config_can_be_downgraded(
+        {'CONFIG_VERSION': {'CURRENT': MAXIMUM_DOWNGRADE_CONFIG_VERSION + 1, 'OLDEST_COMPATIBLE': 0}}
+    )
+
 
 def test_migrate_full(load_config_sample, monkeypatch):
     """Test the full config migration."""