👌 Allow config downgrade for known migrations (#7491)

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.

This change allows to downgrade the config without requiring the current
config schema version be the latest. It only l requires that the
migration from newer to older version is implemented. This allows to
include migrations for older releases through patch releases.

(cherry picked from commit 7f36513)

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

@@ -180,9 +180,10 @@ 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."""
+
+    # Hotfix see issue #7493
+    _can_downgrade: bool = False
 
 
 class ClosedStorage(AiidaException):

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

@@ -34,6 +34,8 @@
 
 CURRENT_CONFIG_VERSION = 9
 OLDEST_COMPATIBLE_CONFIG_VERSION = 9
+# 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')
 
@@ -497,6 +499,41 @@ def get_oldest_compatible_version(config):
     return config.get('CONFIG_VERSION', {}).get('OLDEST_COMPATIBLE', 0)
 
 
+def config_can_be_downgraded(
+    config: ConfigType,
+    target: Optional[int] = 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:
@@ -534,6 +571,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)
@@ -580,13 +624,25 @@ def config_needs_migrating(config, filepath: Optional[str] = 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 supporting up to version {CURRENT_CONFIG_VERSION}. '
-            'Before switching to an older AiiDA version, use a newer AiiDA version that supports '
-            f' at least version {CURRENT_CONFIG_VERSION} and run `verdi config downgrade {CURRENT_CONFIG_VERSION}` '
-            f' to rewrite it for an older version. See {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.'
+            )
+        error = exceptions.ConfigurationVersionError(msg)
+        error._can_downgrade = can_downgrade
+        raise error
 
     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
@@ -209,3 +212,30 @@ def test_config_downgrade(run_cli_command, config_with_profile_factory):
     options = ['config', 'downgrade', '1']
     result = run_cli_command(cmd_verdi.verdi, options, use_subprocess=False)
     assert 'Success: Downgraded' in result.output.strip()
+
+
+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 = run_cli_command(
+        cmd_verdi.verdi, ['config', 'downgrade', '9'], initialize_ctx_obj=False, use_subprocess=False
+    )
+
+    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
+    assert 'broker.task_timeout' not in dictionary['options']

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."""