<fix>[ceph]: ZSTAC-85248 keep snapshot GC failed until backend cleanup

plugin/ceph/src/main/java/org/zstack/storage/ceph/primary/CephPrimaryStorageBase.java

@@ -5594,6 +5594,13 @@ public void success(DeleteSnapshotRsp returnValue) {
 
             @Override
             public void fail(ErrorCode errorCode) {
+                if (!msg.isGcOnFailure()) {
+                    reply.setError(errorCode);
+                    bus.reply(msg, reply);
+                    completion.done();
+                    return;
+                }
+
                 // ceph has trash, so children may not be deleted immediately.
                 DeleteVolumeSnapshotGC snapshotGC = new DeleteVolumeSnapshotGC();
                 snapshotGC.NAME = String.format("gc-ceph-%s-volumesnapshot-path-%s", self.getUuid(), cmd.snapshotPath);

test/src/test/groovy/org/zstack/test/integration/storage/primary/ceph/CephGCCase.groovy

@@ -151,8 +151,10 @@ class CephGCCase extends SubCase {
         } as VolumeSnapshotInventory
 
         def deleteFailed = true
+        def deleteSnapshotCalled = 0
         def deleteSucVol = []
         env.simulator(CephPrimaryStorageBase.DELETE_SNAPSHOT_PATH) { HttpEntity<String> e, EnvSpec spec ->
+            deleteSnapshotCalled++
             def rsp = new CephPrimaryStorageBase.DeleteSnapshotRsp()
             if (deleteFailed) {
                 rsp.setError("it's children in trash, cannot delete")
@@ -207,6 +209,24 @@ class CephGCCase extends SubCase {
         assert volGC.size() == 1
         assert volGC[0].status != GCStatus.Done.toString()
 
+        def deleteSnapshotCalledBeforeTriggerGC = deleteSnapshotCalled
+        def spGCBeforeTriggerGC = queryGCJob {
+            conditions = ["runnerClass=${DeleteVolumeSnapshotGC.class.name}".toString(), "context~=%${sp.primaryStorageInstallPath}%".toString()]
+        } as List<GarbageCollectorInventory>
+
+        triggerGCJob {
+            uuid = spGC[0].uuid
+        }
+        assert deleteSnapshotCalled == deleteSnapshotCalledBeforeTriggerGC + 1
+        assert queryGCJob {
+            conditions = ["runnerClass=${DeleteVolumeSnapshotGC.class.name}".toString(), "context~=%${sp.primaryStorageInstallPath}%".toString()]
+        }.size() == spGCBeforeTriggerGC.size()
+        assert !retryInSecs(3) {
+            queryGCJob {
+                conditions = ["runnerClass=${DeleteVolumeSnapshotGC.class.name}".toString(), "context~=%${sp.primaryStorageInstallPath}%".toString()]
+            }[0].status == GCStatus.Done.toString()
+        }
+
         deleteFailed = false
         triggerGCJob {
             uuid = spGC[0].uuid