Latest backup in any state should not be deleted while partial delete command

ch_backup/ch_backup.py

@@ -162,8 +162,10 @@ def backup(
             ", ".join(map(lambda db: db.name, databases)),
         )
 
+        skip_lock = self._check_schema_only_backup_skip_lock(sources)
+
         try:
-            with self._context.locker():
+            with self._context.locker(disabled=skip_lock):
                 if sources.access:
                     self._access_backup_manager.backup(self._context)
                 if sources.udf:
@@ -332,13 +334,15 @@ def delete(
         found = False
         deleting_backups = []
         retained_backups = []
-        for backup in self._context.backup_layout.get_backups(use_light_meta=True):
+        for i, backup in enumerate(
+            self._context.backup_layout.get_backups(use_light_meta=True)
+        ):
             if backup.name == backup_name:
                 deleting_backups.append(backup)
                 found = True
                 continue
 
-            if purge_partial and backup.state != BackupState.CREATED:
+            if purge_partial and backup.state != BackupState.CREATED and i != 0:
                 deleting_backups.append(backup)
             else:
                 retained_backups.append(backup)
@@ -581,3 +585,13 @@ def _check_min_interval(self, last_backup: BackupMetadata, force: bool) -> bool:
             return True
 
         return False
+
+    def _check_schema_only_backup_skip_lock(self, sources: BackupSources) -> bool:
+        if not sources.schema_only:
+            return False
+
+        skip_lock = self._context.config.get("skip_lock_for_schema_only", None)
+        if not skip_lock:
+            return False
+
+        return skip_lock.get("backup", False)

ch_backup/logic/lock_manager.py

@@ -9,6 +9,7 @@
 
 from kazoo.recipe.lock import Lock
 
+from ch_backup import logging
 from ch_backup.zookeeper.zookeeper import ZookeeperClient, ZookeeperCTL
 
 
@@ -39,22 +40,29 @@ def __init__(self, lock_conf: dict, zk_ctl: Optional[ZookeeperCTL]) -> None:
         self._zk_lock: Optional[Lock] = None
         self._exitcode = lock_conf.get("exitcode")
         self._distributed = zk_ctl is None
+        self._disabled = False
 
-    def __call__(self, distributed: bool = True):  # type: ignore
+    def __call__(self, distributed: bool = True, disabled: bool = False):  # type: ignore
         """
         Call-method for context manager
         """
         self._distributed = distributed
+        self._disabled = disabled
         return self
 
     def __enter__(self):  # type: ignore
         """
         Enter-method for context manager
         """
+        if self._disabled:
+            logging.debug("Lock is disabled on enter.")
+            return self
+
         if self._lock_conf.get("flock"):
             self._flock()
         if self._distributed and self._lock_conf.get("zk_flock"):
             self._zk_flock()
+
         return self
 
     def __exit__(
@@ -66,6 +74,10 @@ def __exit__(
         """
         Exit-method for context manager
         """
+        if self._disabled:
+            logging.debug("Lock is disabled on exit.")
+            return
+
         if self._lock_conf.get("flock"):
             os.close(self._fd)
         if self._zk_client and self._zk_lock is not None:

tests/integration/features/backup_delete.feature

@@ -194,6 +194,42 @@ Feature: Backup & Clean & Restore
       | num | state             | data_count | link_count   | title        |
     And s3 contains 0 objects
 
+  Scenario: Attempt to delete backup with purge partially does not delete latest creating backup
+    When we create clickhouse01 clickhouse backup
+      """
+      schema_only: True
+      """
+    When we create clickhouse01 clickhouse backup
+      """
+      schema_only: True
+      """
+    When we create clickhouse01 clickhouse backup
+      """
+      schema_only: True
+      """
+    And metadata of clickhouse01 backup #0 was adjusted with
+      """
+      meta:
+          state: creating
+      """
+    And metadata of clickhouse01 backup #1 was adjusted with
+      """
+      meta:
+          state: creating
+      """
+    Then we got the following backups on clickhouse01
+      | num | state    | data_count | link_count | title       |
+      | 0   | creating | 0          | 0          | schema-only |
+      | 1   | creating | 0          | 0          | schema-only |
+      | 2   | created  | 0          | 0          | schema-only |
+    When we delete clickhouse01 clickhouse backup #2
+      """
+      purge_partial: true
+      """
+    Then we got the following backups on clickhouse01
+      | num | state    | data_count | link_count | title       |
+      | 0   | creating | 0          | 0          | schema-only |
+
   Scenario: Attempt to delete non-existent backup with force flag succeeds
     When we delete clickhouse01 clickhouse backup "non_existent_backup"
     """

tests/integration/features/lock.feature

@@ -0,0 +1,61 @@
+Feature: Lock
+
+  Background:
+    Given default configuration
+    And a working s3
+    And a working zookeeper on zookeeper01
+    And a working clickhouse on clickhouse01
+    And a working clickhouse on clickhouse02
+
+  @require_version_22.3
+  Scenario: No backup created while zk lock
+    When ch-backup configuration on clickhouse01
+    """
+    backup:
+        skip_lock_for_schema_only:
+            backup: false
+    lock:
+        zk_flock: true
+        zk_flock_path: /ch_backup/zk_flock_path
+    """
+    And on zookeeper01 we create /ch_backup/zk_flock_path
+    When we create clickhouse01 clickhouse backup
+    Then we got no backups on clickhouse01
+
+  @require_version_22.3
+  Scenario: No "schema-only" backup created while zk lock
+    When ch-backup configuration on clickhouse01
+    """
+    backup:
+        skip_lock_for_schema_only:
+            backup: false
+    lock:
+        zk_flock: true
+        zk_flock_path: /ch_backup/zk_flock_path
+    """
+    And on zookeeper01 we create /ch_backup/zk_flock_path
+    When we create clickhouse01 clickhouse backup
+    """
+    schema_only: true
+    """
+    Then we got no backups on clickhouse01
+
+  @require_version_22.3
+  Scenario: Skip zk lock while creating "schema-only" backup
+    When ch-backup configuration on clickhouse01
+    """
+    backup:
+        skip_lock_for_schema_only:
+            backup: true
+    lock:
+        zk_flock: true
+        zk_flock_path: /ch_backup/zk_flock_path
+    """
+    And on zookeeper01 we create /ch_backup/zk_flock_path
+    When we create clickhouse01 clickhouse backup
+    """
+    schema_only: true
+    """
+    Then we got the following backups on clickhouse01
+      | num | state   | schema_only |
+      | 0   | created | True        |

tests/integration/steps/ch_backup.py

@@ -118,6 +118,11 @@ def step_backup_metadata_absent(context, node, backup_id):
     assert_that(backup.meta, not has_entries(expected_meta))
 
 
+@then("we got no backups on {node:w}")
+def step_no_backups(context, node):
+    step_check_backups_conditions(context, node)
+
+
 @then("we got the following backups on {node:w}")
 def step_check_backups_conditions(context, node):
     ch_backup = BackupManager(context, node)