[blob] Use new blob-exists check in single endpoint

Summary:
Despite D13481, multiple `GetItem` DDB calls were still happening inside `service.assign_holder()`. Removed that call there and updated single endpoint to use the new way.

Updated single assign-holder endpoint to use the new batch function.

Depends on D13481

Test Plan:
- Made sure `data_exists` has still correct value for the `POST /blob` endpoint
- Blob integration tests

Reviewers: will, varun, kamil, marcin

Reviewed By: kamil

Subscribers: ashoat, tomek

Differential Revision: https://phab.comm.dev/D13482

services/blob/src/http/handlers/blob.rs

@@ -1,3 +1,5 @@
+use std::collections::HashSet;
+
 use crate::http::errors::handle_blob_service_error;
 use crate::service::BlobService;
 use crate::validate_identifier;
@@ -131,8 +133,15 @@ pub async fn assign_holder_handler(
   validate_identifier!(holder);
   validate_identifier!(blob_hash);
 
-  let data_exists = service.assign_holder(blob_hash, holder).await?;
-  Ok(HttpResponse::Ok().json(web::Json(AssignHolderResponnse { data_exists })))
+  let data_exists = service
+    .find_existing_blobs(HashSet::from([&blob_hash]))
+    .await?
+    .contains(&blob_hash);
+
+  service.assign_holder(blob_hash, holder).await?;
+
+  let response = AssignHolderResponnse { data_exists };
+  Ok(HttpResponse::Ok().json(web::Json(response)))
 }
 
 #[instrument(skip_all, name = "upload_blob", fields(blob_hash))]

services/blob/src/http/handlers/holders.rs

@@ -51,15 +51,15 @@ pub async fn assign_holders_handler(
   let mut results = Vec::with_capacity(requests.len());
   for item in requests {
     let BlobHashAndHolder { blob_hash, holder } = &item;
+    let data_exists = existing_blobs.contains(blob_hash);
     let result = match service.assign_holder(blob_hash, holder).await {
-      Ok(data_exists) => HolderAssignmentResult {
+      Ok(()) => HolderAssignmentResult {
         request: item,
         success: true,
         data_exists,
         holder_already_exists: false,
       },
       Err(BlobServiceError::DB(DBError::ItemAlreadyExists)) => {
-        let data_exists = existing_blobs.contains(blob_hash);
         HolderAssignmentResult {
           request: item,
           success: true,
@@ -69,7 +69,6 @@ pub async fn assign_holders_handler(
       }
       Err(err) => {
         warn!("Holder assignment error: {:?}", err);
-        let data_exists = existing_blobs.contains(blob_hash);
         HolderAssignmentResult {
           request: item,
           success: false,

services/blob/src/service.rs

@@ -230,17 +230,16 @@ impl BlobService {
     &self,
     blob_hash: impl Into<String>,
     holder: impl Into<String>,
-  ) -> BlobServiceResult<bool> {
+  ) -> BlobServiceResult<()> {
     let blob_hash: String = blob_hash.into();
     trace!(blob_hash, "Attempting to assign holder");
     self
       .db
       .put_holder_assignment(blob_hash.clone(), holder.into())
       .await?;
 
-    trace!("Holder assigned. Checking if data exists");
-    let data_exists = self.db.get_blob_item(blob_hash).await?.is_some();
-    Ok(data_exists)
+    trace!("Holder assigned.");
+    Ok(())
   }
 
   pub async fn revoke_holder(