Remove polling for object creation and deletion in S3 and GCS. (#5688)

After uploading (or deleting) an object to S3 or GCS[^azure], we check
in a loop whether the uploaded object exists (or not), until we receive
a positive response. This was done purportedly to work around the
eventual consistency in the storage backend. However, there are some
problems with this:

* GCS, and many known S3 implementations (Amazon S3, [Cloudflare
R2](https://developers.cloudflare.com/r2/reference/consistency/), GCS'
S3 API,
[MinIO](minio/minio#11389 (comment)),
[Oracle
Cloud](https://docs.oracle.com/en-us/iaas/Content/Object/Concepts/objectstorageoverview.htm#features),
[Wasabi](https://docs.wasabi.com/docs/what-data-consistency-model-does-wasabi-employ),
[Ceph](https://openmetal.io/resources/blog/ceph-replication-and-consistency-model-explained/#consistency),
[Backblaze](https://www.backblaze.com/blog/design-thinking-b2-apis-the-hidden-costs-of-s3-compatibility/#:~:text=The%20B2%20architecture%20offers%20what%20one%20could%20consider%20%E2%80%9Cstrong%20consistency.%E2%80%9D))
are now strongly consistent, making this check add an unnecessary
roundtrip.
* [This
paragraph](https://www.allthingsdistributed.com/2021/04/s3-strong-consistency.html#:~:text=For%20S3%2C%20we,copies%20in%202006.)
makes me think that when Amazon S3 was eventually consistent, one client
might have observed that the object exists, but another might not.
Checking that the object is locally visible to the client that uploaded
it, is not very useful, given object storage's distributed nature.
* https://github.com/Netflix/s3mper was made to work around S3's
eventual consistency, but it relied on DynamoDB. I suspect it's not
possible in general to add strong consistency to an eventually
consistent interface, using only that interface.

[^azure]: Not to Azure since the new SDK rewrite.

---
TYPE: IMPROVEMENT
DESC: Reduced latency when uploading objects on S3 and GCS

Author: teo-tsirpanis

tiledb/sm/filesystem/gcs.cc

@@ -394,8 +394,6 @@ void GCS::remove_file(const URI& uri) const {
         "Delete object failed on: " + uri.to_string() + " (" +
         status.message() + ")");
   }
-
-  throw_if_not_ok(wait_for_object_to_be_deleted(bucket_name, object_path));
 }
 
 void GCS::remove_dir(const URI& uri) const {
@@ -744,43 +742,7 @@ Status GCS::copy_object(const URI& old_uri, const URI& new_uri) const {
         status.message() + ")");
   }
 
-  return wait_for_object_to_propagate(new_bucket_name, new_object_path);
-}
-
-Status GCS::wait_for_object_to_propagate(
-    const std::string& bucket_name, const std::string& object_path) const {
-  RETURN_NOT_OK(init_client());
-
-  unsigned attempts = 0;
-  while (attempts++ < constants::gcs_max_attempts) {
-    if (this->is_object(bucket_name, object_path)) {
-      return Status::Ok();
-    }
-
-    std::this_thread::sleep_for(
-        std::chrono::milliseconds(constants::gcs_attempt_sleep_ms));
-  }
-
-  throw GCSException(
-      "Timed out waiting on object to propogate: " + object_path);
-}
-
-Status GCS::wait_for_object_to_be_deleted(
-    const std::string& bucket_name, const std::string& object_path) const {
-  RETURN_NOT_OK(init_client());
-
-  unsigned attempts = 0;
-  while (attempts++ < constants::gcs_max_attempts) {
-    if (!this->is_object(bucket_name, object_path)) {
-      return Status::Ok();
-    }
-
-    std::this_thread::sleep_for(
-        std::chrono::milliseconds(constants::gcs_attempt_sleep_ms));
-  }
-
-  throw GCSException(
-      "Timed out waiting on object to be deleted: " + object_path);
+  return Status::Ok();
 }
 
 Status GCS::wait_for_bucket_to_propagate(const std::string& bucket_name) const {
@@ -1195,23 +1157,8 @@ void GCS::flush(const URI& uri, bool) {
   std::string object_path;
   throw_if_not_ok(parse_gcs_uri(uri, &bucket_name, &object_path));
 
-  // Wait for the last written part to propogate to ensure all parts
-  // are available for composition into a single object.
-  std::string last_part_path = part_paths.back();
-  const Status st = wait_for_object_to_propagate(bucket_name, last_part_path);
-  state->update_st(st);
   state_lck.unlock();
 
-  if (!st.ok()) {
-    // Delete all outstanding part objects.
-    delete_parts(bucket_name, part_paths);
-
-    // Release all instance state associated with this part list
-    // transactions.
-    finish_multi_part_upload(uri);
-    return;
-  }
-
   // Build a list of objects to compose.
   std::vector<google::cloud::storage::ComposeSourceObject> source_objects;
   source_objects.reserve(part_paths.size());
@@ -1252,8 +1199,6 @@ void GCS::flush(const URI& uri, bool) {
         "Compse object failed on: " + uri.to_string() + " (" +
         status.message() + ")");
   }
-
-  throw_if_not_ok(wait_for_object_to_propagate(bucket_name, object_path));
 }
 
 void GCS::delete_parts(
@@ -1334,7 +1279,7 @@ Status GCS::flush_object_direct(const URI& uri) {
         ")");
   }
 
-  return wait_for_object_to_propagate(bucket_name, object_path);
+  return Status::Ok();
 }
 
 uint64_t GCS::read(

tiledb/sm/filesystem/gcs.h

@@ -651,28 +651,6 @@ class GCS : public FilesystemBase {
    */
   Status copy_object(const URI& old_uri, const URI& new_uri) const;
 
-  /**
-   * Waits for a object with `bucket_name` and `object_path`
-   * to exist on GCS.
-   *
-   * @param bucket_name The object's bucket name.
-   * @param object_path The object's path
-   * @return Status
-   */
-  Status wait_for_object_to_propagate(
-      const std::string& bucket_name, const std::string& object_path) const;
-
-  /**
-   * Waits for a object with `bucket_name` and `object_path`
-   * to not exist on GCS.
-   *
-   * @param bucket_name The object's bucket name.
-   * @param object_path The object's path
-   * @return Status
-   */
-  Status wait_for_object_to_be_deleted(
-      const std::string& bucket_name, const std::string& object_path) const;
-
   /**
    * Waits for a bucket with `bucket_name`
    * to exist on GCS.

tiledb/sm/filesystem/s3.cc

@@ -612,9 +612,6 @@ void S3::touch(const URI& uri) const {
         std::string("Cannot touch object '") + uri.c_str() +
         outcome_error_message(put_object_outcome));
   }
-
-  throw_if_not_ok(wait_for_object_to_propagate(
-      put_object_request.GetBucket(), put_object_request.GetKey()));
 }
 
 void S3::write(
@@ -699,9 +696,6 @@ void S3::remove_file(const URI& uri) const {
         std::string("Failed to delete S3 object '") + uri.c_str() +
         outcome_error_message(delete_object_outcome));
   }
-
-  throw_if_not_ok(wait_for_object_to_be_deleted(
-      delete_object_request.GetBucket(), delete_object_request.GetKey()));
 }
 
 std::vector<directory_entry> S3::ls_with_sizes(const URI& parent) const {
@@ -832,8 +826,6 @@ void S3::flush(const URI& uri, bool finalize) {
     // It is safe to unlock the state here
     state_lck.unlock();
 
-    throw_if_not_ok(wait_for_object_to_propagate(move(bucket), move(key)));
-
     throw_if_not_ok(finish_flush_object(std::move(outcome), uri, buff, false));
   } else {
     Aws::S3::Model::AbortMultipartUploadRequest abort_request =
@@ -917,8 +909,6 @@ void S3::finalize_and_flush_object(const URI& uri) {
           std::string("Failed to flush S3 object ") + uri.c_str() +
           outcome_error_message(outcome)};
     }
-
-    throw_if_not_ok(wait_for_object_to_propagate(state.bucket, state.key));
   } else {
     Aws::S3::Model::AbortMultipartUploadRequest abort_request =
         make_multipart_abort_request(state);
@@ -1567,9 +1557,6 @@ Status S3::copy_object(const URI& old_uri, const URI& new_uri) const {
         new_uri.c_str() + outcome_error_message(copy_object_outcome)));
   }
 
-  throw_if_not_ok(wait_for_object_to_propagate(
-      copy_object_request.GetBucket(), copy_object_request.GetKey()));
-
   return Status::Ok();
 }
 
@@ -1660,48 +1647,6 @@ Status S3::initiate_multipart_request(
   return Status::Ok();
 }
 
-Status S3::wait_for_object_to_propagate(
-    const Aws::String& bucket_name, const Aws::String& object_key) const {
-  throw_if_not_ok(init_client());
-
-  unsigned attempts_cnt = 0;
-  while (attempts_cnt++ < constants::s3_max_attempts) {
-    bool exists;
-    RETURN_NOT_OK(is_object(bucket_name, object_key, &exists));
-    if (exists) {
-      return Status::Ok();
-    }
-
-    std::this_thread::sleep_for(
-        std::chrono::milliseconds(constants::s3_attempt_sleep_ms));
-  }
-
-  return LOG_STATUS(Status_S3Error(
-      "Failed waiting for object " +
-      std::string(object_key.c_str(), object_key.size()) + " to be created."));
-}
-
-Status S3::wait_for_object_to_be_deleted(
-    const Aws::String& bucket_name, const Aws::String& object_key) const {
-  throw_if_not_ok(init_client());
-
-  unsigned attempts_cnt = 0;
-  while (attempts_cnt++ < constants::s3_max_attempts) {
-    bool exists;
-    RETURN_NOT_OK(is_object(bucket_name, object_key, &exists));
-    if (!exists) {
-      return Status::Ok();
-    }
-
-    std::this_thread::sleep_for(
-        std::chrono::milliseconds(constants::s3_attempt_sleep_ms));
-  }
-
-  return LOG_STATUS(Status_S3Error(
-      "Failed waiting for object " +
-      std::string(object_key.c_str(), object_key.size()) + " to be deleted."));
-}
-
 Status S3::wait_for_bucket_to_be_created(const URI& bucket_uri) const {
   throw_if_not_ok(init_client());
 
@@ -1821,9 +1766,6 @@ void S3::write_direct(const URI& uri, const void* buffer, uint64_t length) {
         std::string("Cannot write object '") + uri.c_str() +
         outcome_error_message(put_object_outcome));
   }
-
-  throw_if_not_ok(wait_for_object_to_propagate(
-      put_object_request.GetBucket(), put_object_request.GetKey()));
 }
 
 Status S3::write_multipart(

tiledb/sm/filesystem/s3.h

@@ -1436,14 +1436,6 @@ class S3 : public FilesystemBase {
   Status initiate_multipart_request(
       Aws::Http::URI aws_uri, MultiPartUploadState* state);
 
-  /** Waits for the input object to be propagated. */
-  Status wait_for_object_to_propagate(
-      const Aws::String& bucketName, const Aws::String& objectKey) const;
-
-  /** Waits for the input object to be deleted. */
-  Status wait_for_object_to_be_deleted(
-      const Aws::String& bucketName, const Aws::String& objectKey) const;
-
   /** Waits for the bucket to be created. */
   Status wait_for_bucket_to_be_created(const URI& bucket_uri) const;