diff --git a/services/blob/src/database/client.rs b/services/blob/src/database/client.rs
--- a/services/blob/src/database/client.rs
+++ b/services/blob/src/database/client.rs
@@ -88,6 +88,137 @@
       })?;
     Ok(())
   }
+
+  // Inserts a new holder assignment row into the database. Returns Error
+  // if the item already exists or holder format is invalid.
+  pub async fn put_holder_assignment(
+    &self,
+    blob_hash: impl Into<String>,
+    holder: impl Into<String>,
+  ) -> DBResult<()> {
+    let blob_hash: String = blob_hash.into();
+    let holder: String = holder.into();
+
+    validate_holder(&holder)?;
+    let item = HashMap::from([
+      (ATTR_BLOB_HASH.to_string(), AttributeValue::S(blob_hash)),
+      (ATTR_HOLDER.to_string(), AttributeValue::S(holder)),
+      (ATTR_UNCHECKED.to_string(), UncheckedKind::Holder.into()),
+    ]);
+
+    self.insert_item(item).await?;
+    Ok(())
+  }
+
+  /// Deletes a holder assignment row from the table.
+  /// If the blob item for given holder assignment exists, it will be marked as unchecked.
+  ///
+  /// Returns Error if the holder format is invalid or race condition happened.
+  /// Doesn't fail if the holder assignment didn't exist before.
+  pub async fn delete_holder_assignment(
+    &self,
+    blob_hash: impl Into<String>,
+    holder: impl Into<String>,
+  ) -> DBResult<()> {
+    let blob_hash: String = blob_hash.into();
+    let holder: String = holder.into();
+    validate_holder(&holder)?;
+    let mut transaction = Vec::new();
+
+    // delete the holder row
+    let assignment_key = PrimaryKey {
+      blob_hash: blob_hash.clone(),
+      holder: holder.into(),
+    };
+    let delete_request = Delete::builder()
+      .table_name(BLOB_TABLE_NAME)
+      .set_key(Some(assignment_key.into()))
+      .build();
+    transaction
+      .push(TransactWriteItem::builder().delete(delete_request).build());
+
+    // mark the blob item as unchecked if exists
+    let blob_primary_key = PrimaryKey::for_blob_item(blob_hash);
+    if self.get_raw_item(blob_primary_key.clone()).await?.is_some() {
+      let update_request = Update::builder()
+        .table_name(BLOB_TABLE_NAME)
+        .set_key(Some(blob_primary_key.into()))
+        // even though we checked that the blob item exists, we still need to check it again
+        // using DDB built-in conditions in case it was deleted in meantime
+        .condition_expression(
+          "attribute_exists(#blob_hash) AND attribute_exists(#holder)",
+        )
+        .update_expression("SET #unchecked = :unchecked, #last_modified = :now")
+        .expression_attribute_names("#blob_hash", ATTR_BLOB_HASH)
+        .expression_attribute_names("#holder", ATTR_HOLDER)
+        .expression_attribute_names("#unchecked", ATTR_UNCHECKED)
+        .expression_attribute_names("#last_modified", ATTR_LAST_MODIFIED)
+        .expression_attribute_values(":unchecked", UncheckedKind::Blob.into())
+        .expression_attribute_values(
+          ":now",
+          AttributeValue::N(Utc::now().timestamp_millis().to_string()),
+        )
+        .build();
+      transaction
+        .push(TransactWriteItem::builder().update(update_request).build());
+    }
+
+    self
+      .ddb
+      .transact_write_items()
+      .set_transact_items(Some(transaction))
+      .send()
+      .await
+      .map_err(|err| {
+        debug!("DynamoDB client failed to run transaction: {:?}", err);
+        DBError::AwsSdk(err.into())
+      })?;
+    Ok(())
+  }
+
+  /// Queries the table for a list of holders for given blob hash.
+  /// Optionally limits the number of results.
+  pub async fn list_blob_holders(
+    &self,
+    blob_hash: impl Into<String>,
+    limit: Option<i32>,
+  ) -> DBResult<Vec<String>> {
+    let response = self
+      .ddb
+      .query()
+      .table_name(BLOB_TABLE_NAME)
+      .projection_expression("#holder")
+      .key_condition_expression("#blob_hash = :blob_hash")
+      .expression_attribute_names("#blob_hash", ATTR_BLOB_HASH)
+      .expression_attribute_names("#holder", ATTR_HOLDER)
+      .expression_attribute_values(
+        ":blob_hash",
+        AttributeValue::S(blob_hash.into()),
+      )
+      .consistent_read(true)
+      .set_limit(limit)
+      .send()
+      .await
+      .map_err(|err| {
+        error!("DynamoDB client failed to query holders: {:?}", err);
+        DBError::AwsSdk(err.into())
+      })?;
+
+    let Some(items) = response.items else { return Ok(vec![]); };
+    items
+      .into_iter()
+      .filter_map(|mut row| {
+        // filter out rows that are blob items
+        // we cannot do it in key condition expression - it doesn't support the <> operator
+        // filter expression doesn't work either - it doesn't support filtering by sort key
+        match parse_string_attribute(ATTR_HOLDER, row.remove(ATTR_HOLDER)) {
+          Ok(value) if value.as_str() == BLOB_ITEM_ROW_HOLDER_VALUE => None,
+          holder => Some(holder),
+        }
+      })
+      .collect::<Result<Vec<_>, _>>()
+      .map_err(|err| DBError::Attribute(err))
+  }
 }
 
 // private helpers
@@ -154,3 +285,11 @@
       .map(|response| response.item)
   }
 }
+
+fn validate_holder(holder: &str) -> DBResult<()> {
+  if holder == BLOB_ITEM_ROW_HOLDER_VALUE {
+    debug!("Invalid holder: {}", holder);
+    return Err(DBError::Blob(BlobDBError::InvalidInput(holder.to_string())));
+  }
+  Ok(())
+}
diff --git a/services/blob/src/database/errors.rs b/services/blob/src/database/errors.rs
--- a/services/blob/src/database/errors.rs
+++ b/services/blob/src/database/errors.rs
@@ -24,6 +24,7 @@
 pub enum BlobDBError {
   HolderAlreadyExists(String),
   InvalidS3Path(S3PathError),
+  InvalidInput(String),
 }
 
 impl Display for BlobDBError {
@@ -33,6 +34,9 @@
         write!(f, "Item for given holder [{}] already exists", holder)
       }
       BlobDBError::InvalidS3Path(err) => err.fmt(f),
+      BlobDBError::InvalidInput(value) => {
+        write!(f, "Invalid input value [{}]", value)
+      }
     }
   }
 }