rgw: fix collection of object sync errors

Signed-off-by: Yehuda Sadeh <yehuda@redhat.com>

diff --git a/src/rgw/rgw_data_sync.cc b/src/rgw/rgw_data_sync.cc
index dae37b3fe29613c6494da2efbbe2617be1f62fce..f22775b5d620ccc0c5929c7618622e52071168f3 100644 (file)
--- a/src/rgw/rgw_data_sync.cc
+++ b/src/rgw/rgw_data_sync.cc
@@ -2068,6 +2068,7 @@ class RGWBucketSyncSingleEntryCR : public RGWCoroutine {
 
   RGWDataSyncDebugLogger logger;
 
+
 public:
   RGWBucketSyncSingleEntryCR(RGWDataSyncEnv *_sync_env,
                              RGWBucketInfo *_bucket_info,
@@ -2160,9 +2161,9 @@ public:
         yield call(sync_env->error_logger->log_error_cr(sync_env->conn->get_remote_id(), "data", error_ss.str(), retcode, "failed to sync object"));
       }
 done:
-      /* update marker */
-      set_status() << "calling marker_tracker->finish(" << entry_marker << ")";
       if (sync_status == 0) {
+        /* update marker */
+        set_status() << "calling marker_tracker->finish(" << entry_marker << ")";
         yield call(marker_tracker->finish(entry_marker));
         sync_status = retcode;
       }
@@ -2285,7 +2286,9 @@ int RGWBucketShardFullSyncCR::operate()
         }
         while ((int)num_spawned() > spawn_window) {
           yield wait_for_child();
-          while (collect(&ret, lease_stack)) {
+          bool again = true;
+          while (again) {
+            again = collect(&ret, lease_stack);
             if (ret < 0) {
               ldout(sync_env->cct, 0) << "ERROR: a sync operation returned error" << dendl;
               sync_status = ret;
@@ -2297,7 +2300,18 @@ int RGWBucketShardFullSyncCR::operate()
     } while (list_result.is_truncated && sync_status == 0);
     set_status("done iterating over all objects");
     /* wait for all operations to complete */
-    drain_all_but_stack(lease_stack); /* still need to hold lease cr */
+    while ((int)num_spawned() > 1) {
+      yield wait_for_child();
+      bool again = true;
+      while (again) {
+        again = collect(&ret, lease_stack);
+        if (ret < 0) {
+          ldout(sync_env->cct, 0) << "ERROR: a sync operation returned error" << dendl;
+          sync_status = ret;
+          /* we have reported this error */
+        }
+      }
+    }
     /* update sync state to incremental */
     if (sync_status == 0) {
       yield {
@@ -2492,9 +2506,12 @@ int RGWBucketShardIncrementalSyncCR::operate()
           }
           ldout(sync_env->cct, 5) << *this << ": [inc sync] can't do op on key=" << key << " need to wait for conflicting operation to complete" << dendl;
           yield wait_for_child();
-          while (collect(&ret, lease_stack)) {
+          bool again = true;
+          while (again) {
+            again = collect(&ret, lease_stack);
             if (ret < 0) {
               ldout(sync_env->cct, 0) << "ERROR: a child operation returned error (ret=" << ret << ")" << dendl;
+              sync_status = ret;
               /* we have reported this error */
             }
           }
@@ -2524,7 +2541,9 @@ int RGWBucketShardIncrementalSyncCR::operate()
         while ((int)num_spawned() > spawn_window) {
           set_status() << "num_spawned() > spawn_window";
           yield wait_for_child();
-          while (collect(&ret, lease_stack)) {
+          bool again = true;
+          while (again) {
+            again = collect(&ret, lease_stack);
             if (ret < 0) {
               ldout(sync_env->cct, 0) << "ERROR: a sync operation returned error" << dendl;
               sync_status = ret;
@@ -2536,8 +2555,18 @@ int RGWBucketShardIncrementalSyncCR::operate()
       }
     } while (!list_result.empty() && sync_status == 0);
 
-    if (sync_status < 0) {
-      ldout(sync_env->cct, 0) << "ERROR: failure in sync, backing out (sync_status=" << sync_status<< ")" << dendl;
+    while ((int)num_spawned() > 1) {
+      yield wait_for_child();
+      bool again = true;
+      while (again) {
+        again = collect(&ret, lease_stack);
+        if (ret < 0) {
+          ldout(sync_env->cct, 0) << "ERROR: a sync operation returned error" << dendl;
+          sync_status = ret;
+          /* we have reported this error */
+        }
+        /* not waiting for child here */
+      }
     }
 
     yield {
@@ -2545,13 +2574,14 @@ int RGWBucketShardIncrementalSyncCR::operate()
     }
     if (retcode < 0) {
       ldout(sync_env->cct, 0) << "ERROR: marker_tracker->flush() returned retcode=" << retcode << dendl;
-      drain_all_but_stack(lease_stack);
       lease_cr->go_down();
       drain_all();
       return set_cr_error(retcode);
     }
+    if (sync_status < 0) {
+      ldout(sync_env->cct, 0) << "ERROR: failure in sync, backing out (sync_status=" << sync_status<< ")" << dendl;
+    }
 
-    drain_all_but_stack(lease_stack);
     lease_cr->go_down();
     /* wait for all operations to complete */
     drain_all();