diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0a670f4d4..900028294 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,27 @@
+# v0.3.91 (Wed Jul 24 2024)
+
+#### 🐛 Bug Fix
+
+- Fix N query problem with VersionStatusFilter [#1986](https://github.com/dandi/dandi-archive/pull/1986) ([@jjnesbitt](https://github.com/jjnesbitt))
+
+#### Authors: 1
+
+- Jacob Nesbitt ([@jjnesbitt](https://github.com/jjnesbitt))
+
+---
+
+# v0.3.90 (Mon Jul 22 2024)
+
+#### 🐛 Bug Fix
+
+- Automate dandiset unembargo [#1965](https://github.com/dandi/dandi-archive/pull/1965) ([@jjnesbitt](https://github.com/jjnesbitt))
+
+#### Authors: 1
+
+- Jacob Nesbitt ([@jjnesbitt](https://github.com/jjnesbitt))
+
+---
+
 # v0.3.89 (Fri Jul 19 2024)
 
 #### 🐛 Bug Fix
diff --git a/dandiapi/api/admin.py b/dandiapi/api/admin.py
index eec2213d5..0a7060d22 100644
--- a/dandiapi/api/admin.py
+++ b/dandiapi/api/admin.py
@@ -136,12 +136,17 @@ class VersionStatusFilter(admin.SimpleListFilter):
     title = 'status'
     parameter_name = 'status'
 
-    def lookups(self, request, model_admin):
-        qs = model_admin.get_queryset(request)
-        for status in qs.values_list('status', flat=True).distinct():
-            count = qs.filter(status=status).count()
-            if count:
-                yield (status, f'{status} ({count})')
+    def lookups(self, *args, **kwargs):
+        # The queryset for VersionAdmin contains unnecessary data,
+        # so just use base queryset from Version.objects
+        qs = (
+            Version.objects.values_list('status')
+            .distinct()
+            .annotate(total=Count('status'))
+            .order_by()
+        )
+        for status, count in qs:
+            yield (status, f'{status} ({count})')
 
     def queryset(self, request, queryset):
         status = self.value()
diff --git a/dandiapi/api/mail.py b/dandiapi/api/mail.py
index d68395d69..7b0a5b4d1 100644
--- a/dandiapi/api/mail.py
+++ b/dandiapi/api/mail.py
@@ -43,8 +43,18 @@ def user_greeting_name(user: User, socialaccount: SocialAccount = None) -> str:
     return social_user['username']
 
 
-def build_message(subject: str, message: str, to: list[str], html_message: str | None = None):
-    email_message = mail.EmailMultiAlternatives(subject=subject, body=message, to=to)
+def build_message(  # noqa: PLR0913
+    to: list[str],
+    subject: str,
+    message: str,
+    html_message: str | None = None,
+    cc: list[str] | None = None,
+    bcc: list[str] | None = None,
+    reply_to: list[str] | None = None,
+):
+    email_message = mail.EmailMultiAlternatives(
+        subject=subject, body=message, to=to, cc=cc, bcc=bcc, reply_to=reply_to
+    )
     if html_message is not None:
         email_message.attach_alternative(html_message, 'text/html')
     return email_message
@@ -182,31 +192,6 @@ def send_pending_users_message(users: Iterable[User]):
         connection.send_messages(messages)
 
 
-def build_dandisets_to_unembargo_message(dandisets: Iterable[Dandiset]):
-    dandiset_context = [
-        {
-            'identifier': ds.identifier,
-            'owners': [user.username for user in ds.owners],
-            'asset_count': ds.draft_version.asset_count,
-            'size': ds.draft_version.size,
-        }
-        for ds in dandisets
-    ]
-    render_context = {**BASE_RENDER_CONTEXT, 'dandisets': dandiset_context}
-    return build_message(
-        subject='DANDI: New Dandisets to unembargo',
-        message=render_to_string('api/mail/dandisets_to_unembargo.txt', render_context),
-        to=[settings.DANDI_DEV_EMAIL],
-    )
-
-
-def send_dandisets_to_unembargo_message(dandisets: Iterable[Dandiset]):
-    logger.info('Sending dandisets to unembargo message to devs at %s', settings.DANDI_DEV_EMAIL)
-    messages = [build_dandisets_to_unembargo_message(dandisets)]
-    with mail.get_connection() as connection:
-        connection.send_messages(messages)
-
-
 def build_dandiset_unembargoed_message(dandiset: Dandiset):
     dandiset_context = {
         'identifier': dandiset.identifier,
@@ -231,3 +216,30 @@ def send_dandiset_unembargoed_message(dandiset: Dandiset):
     messages = [build_dandiset_unembargoed_message(dandiset)]
     with mail.get_connection() as connection:
         connection.send_messages(messages)
+
+
+def build_dandiset_unembargo_failed_message(dandiset: Dandiset):
+    dandiset_context = {
+        'identifier': dandiset.identifier,
+    }
+
+    render_context = {**BASE_RENDER_CONTEXT, 'dandiset': dandiset_context}
+    html_message = render_to_string('api/mail/dandiset_unembargo_failed.html', render_context)
+    return build_message(
+        subject=f'DANDI: Unembargo failed for dandiset {dandiset.identifier}',
+        message=strip_tags(html_message),
+        html_message=html_message,
+        to=[owner.email for owner in dandiset.owners],
+        bcc=[settings.DANDI_DEV_EMAIL],
+        reply_to=[ADMIN_EMAIL],
+    )
+
+
+def send_dandiset_unembargo_failed_message(dandiset: Dandiset):
+    logger.info(
+        'Sending dandiset unembargo failed message for dandiset %s to dandiset owners and devs',
+        dandiset.identifier,
+    )
+    messages = [build_dandiset_unembargo_failed_message(dandiset)]
+    with mail.get_connection() as connection:
+        connection.send_messages(messages)
diff --git a/dandiapi/api/models/dandiset.py b/dandiapi/api/models/dandiset.py
index 595467dc7..4d95b6db4 100644
--- a/dandiapi/api/models/dandiset.py
+++ b/dandiapi/api/models/dandiset.py
@@ -62,6 +62,10 @@ def identifier(self) -> str:
     def embargoed(self) -> bool:
         return self.embargo_status == self.EmbargoStatus.EMBARGOED
 
+    @property
+    def unembargo_in_progress(self) -> bool:
+        return self.embargo_status == self.EmbargoStatus.UNEMBARGOING
+
     @property
     def most_recent_published_version(self):
         return self.versions.exclude(version='draft').order_by('modified').last()
diff --git a/dandiapi/api/services/dandiset/__init__.py b/dandiapi/api/services/dandiset/__init__.py
index 6060ab00e..f3016734b 100644
--- a/dandiapi/api/services/dandiset/__init__.py
+++ b/dandiapi/api/services/dandiset/__init__.py
@@ -62,7 +62,7 @@ def delete_dandiset(*, user, dandiset: Dandiset) -> None:
         raise NotAllowedError('Cannot delete dandisets with published versions.')
     if dandiset.versions.filter(status=Version.Status.PUBLISHING).exists():
         raise NotAllowedError('Cannot delete dandisets that are currently being published.')
-    if dandiset.embargo_status == Dandiset.EmbargoStatus.UNEMBARGOING:
+    if dandiset.unembargo_in_progress:
         raise DandisetUnembargoInProgressError
 
     # Delete all versions first, so that AssetPath deletion is cascaded
diff --git a/dandiapi/api/services/embargo/__init__.py b/dandiapi/api/services/embargo/__init__.py
index a7b66a7d6..ffb7ff71e 100644
--- a/dandiapi/api/services/embargo/__init__.py
+++ b/dandiapi/api/services/embargo/__init__.py
@@ -1,30 +1,37 @@
 from __future__ import annotations
 
 from concurrent.futures import ThreadPoolExecutor
+import logging
 from typing import TYPE_CHECKING
 
 from botocore.config import Config
 from django.conf import settings
 from django.db import transaction
+from more_itertools import chunked
 
-from dandiapi.api.copy import copy_object_multipart
-from dandiapi.api.mail import send_dandisets_to_unembargo_message
-from dandiapi.api.models import Asset, AssetBlob, AuditRecord, Dandiset, Upload, Version
+from dandiapi.api.mail import send_dandiset_unembargoed_message
+from dandiapi.api.models import Asset, AssetBlob, AuditRecord, Dandiset, Version
 from dandiapi.api.services.asset.exceptions import DandisetOwnerRequiredError
+from dandiapi.api.services.exceptions import DandiError
 from dandiapi.api.storage import get_boto_client
+from dandiapi.api.tasks import unembargo_dandiset_task
 
 from .exceptions import (
     AssetBlobEmbargoedError,
+    AssetTagRemovalError,
     DandisetActiveUploadsError,
     DandisetNotEmbargoedError,
 )
 
 if TYPE_CHECKING:
     from django.contrib.auth.models import User
-    from django.db.models import QuerySet
     from mypy_boto3_s3 import S3Client
 
 
+logger = logging.getLogger(__name__)
+ASSET_BLOB_TAG_REMOVAL_CHUNK_SIZE = 5000
+
+
 def _delete_asset_blob_tags(client: S3Client, blob: str):
     client.delete_object_tagging(
         Bucket=settings.DANDI_DANDISETS_BUCKET_NAME,
@@ -34,28 +41,66 @@ def _delete_asset_blob_tags(client: S3Client, blob: str):
 
 # NOTE: In testing this took ~2 minutes for 100,000 files
 def _remove_dandiset_asset_blob_embargo_tags(dandiset: Dandiset):
-    # First we need to generate a CSV manifest containing all asset blobs that need to be untaged
-    embargoed_asset_blobs = AssetBlob.objects.filter(
-        embargoed=True, assets__versions__dandiset=dandiset
-    ).values_list('blob', flat=True)
-
     client = get_boto_client(config=Config(max_pool_connections=100))
-    with ThreadPoolExecutor(max_workers=100) as e:
-        for blob in embargoed_asset_blobs:
-            e.submit(_delete_asset_blob_tags, client=client, blob=blob)
+    embargoed_asset_blobs = (
+        AssetBlob.objects.filter(embargoed=True, assets__versions__dandiset=dandiset)
+        .values_list('blob', flat=True)
+        .iterator(chunk_size=ASSET_BLOB_TAG_REMOVAL_CHUNK_SIZE)
+    )
+
+    # Chunk the blobs so we're never storing a list of all embargoed blobs
+    chunks = chunked(embargoed_asset_blobs, ASSET_BLOB_TAG_REMOVAL_CHUNK_SIZE)
+    for chunk in chunks:
+        with ThreadPoolExecutor(max_workers=100) as e:
+            futures = [
+                e.submit(_delete_asset_blob_tags, client=client, blob=blob) for blob in chunk
+            ]
+
+        # Check if any failed and raise exception if so
+        failed = [blob for i, blob in enumerate(chunk) if futures[i].exception() is not None]
+        if failed:
+            raise AssetTagRemovalError('Some blobs failed to remove tags', blobs=failed)
 
 
 @transaction.atomic()
-def _unembargo_dandiset(dandiset: Dandiset, user: User):
-    # NOTE: Before proceeding, all asset blobs must have their embargoed tags removed in s3
+def unembargo_dandiset(ds: Dandiset, user: User):
+    """Unembargo a dandiset by copying all embargoed asset blobs to the public bucket."""
+    logger.info('Unembargoing Dandiset %s', ds.identifier)
+    logger.info('\t%s assets', ds.draft_version.assets.count())
+
+    if ds.embargo_status != Dandiset.EmbargoStatus.UNEMBARGOING:
+        raise DandiError(
+            message=f'Expected dandiset state {Dandiset.EmbargoStatus.UNEMBARGOING}, found {ds.embargo_status}',  # noqa: E501
+            http_status_code=500,
+        )
+    if ds.uploads.all().exists():
+        raise DandisetActiveUploadsError(http_status_code=500)
+
+    # Remove tags in S3
+    logger.info('Removing tags...')
+    _remove_dandiset_asset_blob_embargo_tags(ds)
+
+    # Update embargoed flag on asset blobs
+    updated = AssetBlob.objects.filter(embargoed=True, assets__versions__dandiset=ds).update(
+        embargoed=False
+    )
+    logger.info('Updated %s asset blobs', updated)
+
+    # Set status to OPEN
+    Dandiset.objects.filter(pk=ds.pk).update(embargo_status=Dandiset.EmbargoStatus.OPEN)
+    logger.info('Dandiset embargo status updated')
 
-    draft_version: Version = dandiset.draft_version
-    embargoed_assets: QuerySet[Asset] = draft_version.assets.filter(blob__embargoed=True)
-    AssetBlob.objects.filter(assets__in=embargoed_assets).update(embargoed=False)
+    # Fetch version to ensure changed embargo_status is included
+    # Save version to update metadata through populate_metadata
+    v = Version.objects.select_for_update().get(dandiset=ds, version='draft')
+    v.save()
+    logger.info('Version metadata updated')
 
-    # Set access on dandiset
-    dandiset.embargo_status = Dandiset.EmbargoStatus.OPEN
-    dandiset.save()
+    # Notify owners of completed unembargo
+    send_dandiset_unembargoed_message(ds)
+    logger.info('Dandiset owners notified.')
+
+    logger.info('...Done')
 
     audit_record = AuditRecord.unembargo_dandiset(dandiset=dandiset, user=user)
     audit_record.save()
@@ -69,8 +114,8 @@ def remove_asset_blob_embargoed_tag(asset_blob: AssetBlob) -> None:
     _delete_asset_blob_tags(client=get_boto_client(), blob=asset_blob.blob.name)
 
 
-def unembargo_dandiset(*, user: User, dandiset: Dandiset):
-    """Unembargo a dandiset by copying all embargoed asset blobs to the public bucket."""
+def kickoff_dandiset_unembargo(*, user: User, dandiset: Dandiset):
+    """Set dandiset status to kickoff unembargo."""
     if dandiset.embargo_status != Dandiset.EmbargoStatus.EMBARGOED:
         raise DandisetNotEmbargoedError
 
@@ -81,10 +126,8 @@ def unembargo_dandiset(*, user: User, dandiset: Dandiset):
     if dandiset.uploads.count():
         raise DandisetActiveUploadsError
 
-    # A scheduled task will pick up any new dandisets with this status and email the admins to
-    # initiate the unembargo process
-    dandiset.embargo_status = Dandiset.EmbargoStatus.UNEMBARGOING
-    dandiset.save()
-
-    # Send initial email to ensure it's seen in a timely manner
-    send_dandisets_to_unembargo_message(dandisets=[dandiset])
+    with transaction.atomic():
+        Dandiset.objects.filter(pk=dandiset.pk).update(
+            embargo_status=Dandiset.EmbargoStatus.UNEMBARGOING
+        )
+        transaction.on_commit(lambda: unembargo_dandiset_task.delay(dandiset.pk))
diff --git a/dandiapi/api/services/embargo/exceptions.py b/dandiapi/api/services/embargo/exceptions.py
index 13038bd59..1bdc0fb25 100644
--- a/dandiapi/api/services/embargo/exceptions.py
+++ b/dandiapi/api/services/embargo/exceptions.py
@@ -30,3 +30,9 @@ class UnauthorizedEmbargoAccessError(DandiError):
     message = (
         'Authentication credentials must be provided when attempting to access embargoed dandisets'
     )
+
+
+class AssetTagRemovalError(Exception):
+    def __init__(self, message: str, blobs: list[str]) -> None:
+        super().__init__(message)
+        self.blobs = blobs
diff --git a/dandiapi/api/tasks/__init__.py b/dandiapi/api/tasks/__init__.py
index ac743bef8..3ae3c7ebd 100644
--- a/dandiapi/api/tasks/__init__.py
+++ b/dandiapi/api/tasks/__init__.py
@@ -5,6 +5,7 @@
 from django.contrib.auth.models import User
 
 from dandiapi.api.doi import delete_doi
+from dandiapi.api.mail import send_dandiset_unembargo_failed_message
 from dandiapi.api.manifests import (
     write_assets_jsonld,
     write_assets_yaml,
@@ -13,6 +14,7 @@
     write_dandiset_yaml,
 )
 from dandiapi.api.models import Asset, AssetBlob, Version
+from dandiapi.api.models.dandiset import Dandiset
 
 logger = get_task_logger(__name__)
 
@@ -84,3 +86,17 @@ def publish_dandiset_task(dandiset_id: int, user_id: int):
     from dandiapi.api.services.publish import _publish_dandiset
 
     _publish_dandiset(dandiset_id=dandiset_id, user_id=user_id)
+
+
+@shared_task(soft_time_limit=1200)
+def unembargo_dandiset_task(dandiset_id: int):
+    from dandiapi.api.services.embargo import unembargo_dandiset
+
+    ds = Dandiset.objects.get(pk=dandiset_id)
+
+    # If the unembargo fails for any reason, send an email, but continue the error propagation
+    try:
+        unembargo_dandiset(ds, user)
+    except Exception:
+        send_dandiset_unembargo_failed_message(ds)
+        raise
diff --git a/dandiapi/api/tasks/scheduled.py b/dandiapi/api/tasks/scheduled.py
index 9fa8af71d..2b4a100e9 100644
--- a/dandiapi/api/tasks/scheduled.py
+++ b/dandiapi/api/tasks/scheduled.py
@@ -20,10 +20,9 @@
 from django.db.models.query_utils import Q
 
 from dandiapi.analytics.tasks import collect_s3_log_records_task
-from dandiapi.api.mail import send_dandisets_to_unembargo_message, send_pending_users_message
+from dandiapi.api.mail import send_pending_users_message
 from dandiapi.api.models import UserMetadata, Version
 from dandiapi.api.models.asset import Asset
-from dandiapi.api.models.dandiset import Dandiset
 from dandiapi.api.services.metadata import version_aggregate_assets_summary
 from dandiapi.api.services.metadata.exceptions import VersionMetadataConcurrentlyModifiedError
 from dandiapi.api.tasks import (
@@ -112,14 +111,6 @@ def send_pending_users_email() -> None:
         send_pending_users_message(pending_users)
 
 
-@shared_task(soft_time_limit=20)
-def send_dandisets_to_unembargo_email() -> None:
-    """Send an email to admins listing dandisets that have requested unembargo."""
-    dandisets = Dandiset.objects.filter(embargo_status=Dandiset.EmbargoStatus.UNEMBARGOING)
-    if dandisets.exists():
-        send_dandisets_to_unembargo_message(dandisets)
-
-
 @shared_task(soft_time_limit=60)
 def refresh_materialized_view_search() -> None:
     """
@@ -148,9 +139,6 @@ def register_scheduled_tasks(sender: Celery, **kwargs):
     # Send daily email to admins containing a list of users awaiting approval
     sender.add_periodic_task(crontab(hour=0, minute=0), send_pending_users_email.s())
 
-    # Send daily email to admins containing a list of dandisets to unembargo
-    sender.add_periodic_task(crontab(hour=0, minute=0), send_dandisets_to_unembargo_email.s())
-
     # Refresh the materialized view used by asset search every 10 mins.
     sender.add_periodic_task(timedelta(minutes=10), refresh_materialized_view_search.s())
 
diff --git a/dandiapi/api/templates/api/mail/dandiset_unembargo_failed.html b/dandiapi/api/templates/api/mail/dandiset_unembargo_failed.html
new file mode 100644
index 000000000..7a4b1356c
--- /dev/null
+++ b/dandiapi/api/templates/api/mail/dandiset_unembargo_failed.html
@@ -0,0 +1,7 @@
+There was an error during the unembargo of dandiset
+<a href="{{ dandiset.ui_link }}">{{dandiset.identifier }}</a>. This has been reported to the developers, and will be
+investigated as soon as possible. We hope to have this resolved soon! Please avoid making any major changes to this
+dandiset in the meantime, as that may hinder or delay the process of resolving this issue.
+If in two business days your dandiset is still embargoed, please reply to this email.
+<br><br><br>
+<em>You are receiving this email because you are an owner of this dandiset.</em>
\ No newline at end of file
diff --git a/dandiapi/api/templates/api/mail/dandisets_to_unembargo.txt b/dandiapi/api/templates/api/mail/dandisets_to_unembargo.txt
deleted file mode 100644
index c716839b1..000000000
--- a/dandiapi/api/templates/api/mail/dandisets_to_unembargo.txt
+++ /dev/null
@@ -1,10 +0,0 @@
-{% autoescape off %}
-The following new dandisets are awaiting unembargo:
-
-{% for ds in dandisets %}
-Dandiset ID: {{ ds.identifier }}
-Owners: {{ ds.owners }}
-Number of Assets: {{ ds.asset_count }}
-Total data size: {{ ds.size }}
-{% endfor %}
-{% endautoescape %}
diff --git a/dandiapi/api/tests/test_asset.py b/dandiapi/api/tests/test_asset.py
index 68dadca42..fc0e32f4a 100644
--- a/dandiapi/api/tests/test_asset.py
+++ b/dandiapi/api/tests/test_asset.py
@@ -637,7 +637,7 @@ def test_asset_create_embargo(
 
 
 @pytest.mark.django_db()
-def test_asset_create_unembargoing(
+def test_asset_create_unembargo_in_progress(
     api_client, user, draft_version_factory, dandiset_factory, embargoed_asset_blob
 ):
     dandiset = dandiset_factory(embargo_status=Dandiset.EmbargoStatus.UNEMBARGOING)
@@ -1122,7 +1122,7 @@ def test_asset_rest_update_embargo(api_client, user, draft_version, asset, embar
 
 
 @pytest.mark.django_db()
-def test_asset_rest_update_unembargoing(
+def test_asset_rest_update_unembargo_in_progress(
     api_client, user, draft_version_factory, asset, embargoed_asset_blob
 ):
     draft_version = draft_version_factory(
@@ -1325,7 +1325,7 @@ def test_asset_rest_delete(api_client, user, draft_version, asset):
 
 
 @pytest.mark.django_db()
-def test_asset_rest_delete_unembargoing(api_client, user, draft_version_factory, asset):
+def test_asset_rest_delete_unembargo_in_progress(api_client, user, draft_version_factory, asset):
     draft_version = draft_version_factory(
         dandiset__embargo_status=Dandiset.EmbargoStatus.UNEMBARGOING
     )
diff --git a/dandiapi/api/tests/test_dandiset.py b/dandiapi/api/tests/test_dandiset.py
index 9efeed612..4ac6d78e1 100644
--- a/dandiapi/api/tests/test_dandiset.py
+++ b/dandiapi/api/tests/test_dandiset.py
@@ -735,30 +735,28 @@ def test_dandiset_rest_create_with_invalid_identifier(api_client, admin_user):
 
 
 @pytest.mark.django_db()
-def test_dandiset_rest_delete(api_client, draft_version_factory, user):
+@pytest.mark.parametrize(
+    ('embargo_status', 'success'),
+    [
+        (Dandiset.EmbargoStatus.OPEN, True),
+        (Dandiset.EmbargoStatus.EMBARGOED, True),
+        (Dandiset.EmbargoStatus.UNEMBARGOING, False),
+    ],
+)
+def test_dandiset_rest_delete(api_client, draft_version_factory, user, embargo_status, success):
     api_client.force_authenticate(user=user)
 
     # Ensure that open or embargoed dandisets can be deleted
-    draft_version = draft_version_factory(dandiset__embargo_status=Dandiset.EmbargoStatus.OPEN)
-    assign_perm('owner', user, draft_version.dandiset)
-    response = api_client.delete(f'/api/dandisets/{draft_version.dandiset.identifier}/')
-    assert response.status_code == 204
-    assert not Dandiset.objects.all()
-
-    draft_version = draft_version_factory(dandiset__embargo_status=Dandiset.EmbargoStatus.EMBARGOED)
+    draft_version = draft_version_factory(dandiset__embargo_status=embargo_status)
     assign_perm('owner', user, draft_version.dandiset)
     response = api_client.delete(f'/api/dandisets/{draft_version.dandiset.identifier}/')
-    assert response.status_code == 204
-    assert not Dandiset.objects.all()
 
-    # Ensure that currently unembargoing dandisets can't be deleted
-    draft_version = draft_version_factory(
-        dandiset__embargo_status=Dandiset.EmbargoStatus.UNEMBARGOING
-    )
-    assign_perm('owner', user, draft_version.dandiset)
-    response = api_client.delete(f'/api/dandisets/{draft_version.dandiset.identifier}/')
-    assert response.status_code == 400
-    assert Dandiset.objects.count() == 1
+    if success:
+        assert response.status_code == 204
+        assert not Dandiset.objects.all()
+    else:
+        assert response.status_code >= 400
+        assert Dandiset.objects.count() == 1
 
 
 @pytest.mark.django_db()
@@ -885,13 +883,13 @@ def test_dandiset_rest_change_owner(
 
 
 @pytest.mark.django_db()
-def test_dandiset_rest_change_owners_unembargoing(
+def test_dandiset_rest_change_owners_unembargo_in_progress(
     api_client,
     draft_version_factory,
     user_factory,
     social_account_factory,
 ):
-    """Test that unembargoing a dandiset prevents user modification."""
+    """Test that a dandiset undergoing unembargo prevents user modification."""
     draft_version = draft_version_factory(
         dandiset__embargo_status=Dandiset.EmbargoStatus.UNEMBARGOING
     )
diff --git a/dandiapi/api/tests/test_unembargo.py b/dandiapi/api/tests/test_unembargo.py
index b4976d832..0d6eff3e8 100644
--- a/dandiapi/api/tests/test_unembargo.py
+++ b/dandiapi/api/tests/test_unembargo.py
@@ -1,11 +1,28 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
+import dandischema
 from guardian.shortcuts import assign_perm
 import pytest
 
 from dandiapi.api.models.dandiset import Dandiset
-from dandiapi.api.services.embargo import AssetBlobEmbargoedError, remove_asset_blob_embargoed_tag
-from dandiapi.api.tasks.scheduled import send_dandisets_to_unembargo_email
+from dandiapi.api.services.embargo import (
+    AssetBlobEmbargoedError,
+    _remove_dandiset_asset_blob_embargo_tags,
+    remove_asset_blob_embargoed_tag,
+    unembargo_dandiset,
+)
+from dandiapi.api.services.embargo.exceptions import (
+    AssetTagRemovalError,
+    DandisetActiveUploadsError,
+)
+from dandiapi.api.services.exceptions import DandiError
+from dandiapi.api.tasks import unembargo_dandiset_task
+
+if TYPE_CHECKING:
+    from dandiapi.api.models.asset import AssetBlob
+    from dandiapi.api.models.version import Version
 
 
 @pytest.mark.django_db()
@@ -18,39 +35,199 @@ def test_remove_asset_blob_embargoed_tag_fails_on_embargod(embargoed_asset_blob,
 
 
 @pytest.mark.django_db()
-def test_unembargo_dandiset_sends_emails(
-    api_client, user, dandiset, draft_version_factory, mailoutbox
+def test_kickoff_dandiset_unembargo_dandiset_not_embargoed(
+    api_client, user, dandiset_factory, draft_version_factory
 ):
+    dandiset = dandiset_factory(embargo_status=Dandiset.EmbargoStatus.OPEN)
     draft_version_factory(dandiset=dandiset)
-
     assign_perm('owner', user, dandiset)
     api_client.force_authenticate(user=user)
 
-    dandiset.embargo_status = Dandiset.EmbargoStatus.EMBARGOED
-    dandiset.save()
-
     resp = api_client.post(f'/api/dandisets/{dandiset.identifier}/unembargo/')
-    assert resp.status_code == 200
+    assert resp.status_code == 400
 
-    # Simulate the scheduled task calling this function
-    send_dandisets_to_unembargo_email()
 
-    assert mailoutbox
-    assert 'unembargo' in mailoutbox[0].subject
-    assert dandiset.identifier in mailoutbox[0].message().get_payload()
-    assert user.username in mailoutbox[0].message().get_payload()
+@pytest.mark.django_db()
+def test_kickoff_dandiset_unembargo_not_owner(
+    api_client, user, dandiset_factory, draft_version_factory
+):
+    dandiset = dandiset_factory(embargo_status=Dandiset.EmbargoStatus.EMBARGOED)
+    draft_version_factory(dandiset=dandiset)
+    api_client.force_authenticate(user=user)
+
+    resp = api_client.post(f'/api/dandisets/{dandiset.identifier}/unembargo/')
+    assert resp.status_code == 403
 
 
 @pytest.mark.django_db()
-def test_unembargo_dandiset_lingering_uploads(
+def test_kickoff_dandiset_unembargo_active_uploads(
     api_client, user, dandiset_factory, draft_version_factory, upload_factory
 ):
     dandiset = dandiset_factory(embargo_status=Dandiset.EmbargoStatus.EMBARGOED)
     draft_version_factory(dandiset=dandiset)
-    upload_factory(dandiset=dandiset)
-
     assign_perm('owner', user, dandiset)
     api_client.force_authenticate(user=user)
 
+    # Test that active uploads prevent unembargp
+    upload_factory(dandiset=dandiset)
     resp = api_client.post(f'/api/dandisets/{dandiset.identifier}/unembargo/')
     assert resp.status_code == 400
+
+
+# transaction=True required due to how `kickoff_dandiset_unembargo` calls `unembargo_dandiset_task`
+@pytest.mark.django_db(transaction=True)
+def test_kickoff_dandiset_unembargo(api_client, user, draft_version_factory, mailoutbox, mocker):
+    draft_version = draft_version_factory(dandiset__embargo_status=Dandiset.EmbargoStatus.EMBARGOED)
+    ds: Dandiset = draft_version.dandiset
+
+    assign_perm('owner', user, ds)
+    api_client.force_authenticate(user=user)
+
+    # mock this task to check if called
+    patched_task = mocker.patch('dandiapi.api.services.embargo.unembargo_dandiset_task')
+
+    resp = api_client.post(f'/api/dandisets/{ds.identifier}/unembargo/')
+    assert resp.status_code == 200
+
+    ds.refresh_from_db()
+    assert ds.embargo_status == Dandiset.EmbargoStatus.UNEMBARGOING
+
+    # Check that unembargo dandiset task was delayed
+    assert len(patched_task.mock_calls) == 1
+    assert str(patched_task.mock_calls[0]) == f'call.delay({ds.pk})'
+
+
+@pytest.mark.django_db()
+def test_unembargo_dandiset_not_unembargoing(draft_version_factory):
+    draft_version = draft_version_factory(dandiset__embargo_status=Dandiset.EmbargoStatus.EMBARGOED)
+    ds: Dandiset = draft_version.dandiset
+
+    with pytest.raises(DandiError):
+        unembargo_dandiset(ds)
+
+
+@pytest.mark.django_db()
+def test_unembargo_dandiset_uploads_exist(draft_version_factory, upload_factory):
+    draft_version = draft_version_factory(
+        dandiset__embargo_status=Dandiset.EmbargoStatus.UNEMBARGOING
+    )
+    ds: Dandiset = draft_version.dandiset
+
+    upload_factory(dandiset=ds)
+    with pytest.raises(DandisetActiveUploadsError):
+        unembargo_dandiset(ds)
+
+
+@pytest.mark.django_db()
+def test_remove_dandiset_asset_blob_embargo_tags_chunks(
+    draft_version_factory,
+    asset_factory,
+    embargoed_asset_blob_factory,
+    mocker,
+):
+    delete_asset_blob_tags_mock = mocker.patch(
+        'dandiapi.api.services.embargo._delete_asset_blob_tags'
+    )
+    chunk_size = mocker.patch('dandiapi.api.services.embargo.ASSET_BLOB_TAG_REMOVAL_CHUNK_SIZE', 2)
+
+    draft_version: Version = draft_version_factory(
+        dandiset__embargo_status=Dandiset.EmbargoStatus.UNEMBARGOING
+    )
+    ds: Dandiset = draft_version.dandiset
+    for _ in range(chunk_size + 1):
+        asset = asset_factory(blob=embargoed_asset_blob_factory())
+        draft_version.assets.add(asset)
+
+    _remove_dandiset_asset_blob_embargo_tags(dandiset=ds)
+
+    # Assert that _delete_asset_blob_tags was called chunk_size +1 times, to ensure that it works
+    # correctly across chunks
+    assert len(delete_asset_blob_tags_mock.mock_calls) == chunk_size + 1
+
+
+@pytest.mark.django_db()
+def test_delete_asset_blob_tags_fails(
+    draft_version_factory,
+    asset_factory,
+    embargoed_asset_blob_factory,
+    mocker,
+):
+    mocker.patch('dandiapi.api.services.embargo._delete_asset_blob_tags', side_effect=ValueError)
+    draft_version: Version = draft_version_factory(
+        dandiset__embargo_status=Dandiset.EmbargoStatus.UNEMBARGOING
+    )
+    ds: Dandiset = draft_version.dandiset
+    asset = asset_factory(blob=embargoed_asset_blob_factory())
+    draft_version.assets.add(asset)
+
+    # Check that if an exception within `_delete_asset_blob_tags` is raised, it's propagated upwards
+    # as an AssetTagRemovalError
+    with pytest.raises(AssetTagRemovalError):
+        _remove_dandiset_asset_blob_embargo_tags(dandiset=ds)
+
+
+@pytest.mark.django_db()
+def test_unembargo_dandiset(
+    draft_version_factory,
+    asset_factory,
+    embargoed_asset_blob_factory,
+    mocker,
+    mailoutbox,
+    user_factory,
+):
+    draft_version: Version = draft_version_factory(
+        dandiset__embargo_status=Dandiset.EmbargoStatus.UNEMBARGOING
+    )
+    ds: Dandiset = draft_version.dandiset
+    owners = [user_factory() for _ in range(5)]
+    for user in owners:
+        assign_perm('owner', user, ds)
+
+    embargoed_blob: AssetBlob = embargoed_asset_blob_factory()
+    asset = asset_factory(blob=embargoed_blob)
+    draft_version.assets.add(asset)
+    assert embargoed_blob.embargoed
+
+    # Patch this function to check if it's been called, since we can't test the tagging directly
+    patched = mocker.patch('dandiapi.api.services.embargo._delete_asset_blob_tags')
+
+    unembargo_dandiset(ds)
+    patched.assert_called_once()
+
+    embargoed_blob.refresh_from_db()
+    ds.refresh_from_db()
+    draft_version.refresh_from_db()
+    assert not embargoed_blob.embargoed
+    assert ds.embargo_status == Dandiset.EmbargoStatus.OPEN
+    assert (
+        draft_version.metadata['access'][0]['status']
+        == dandischema.models.AccessType.OpenAccess.value
+    )
+
+    # Check that a correct email exists
+    assert mailoutbox
+    assert 'has been unembargoed' in mailoutbox[0].subject
+    payload = mailoutbox[0].message().get_payload()[0].get_payload()
+    assert ds.identifier in payload
+    assert 'has been unembargoed' in payload
+
+    # Check that the email was sent to all owners
+    owner_email_set = {user.email for user in owners}
+    mailoutbox_to_email_set = set(mailoutbox[0].to)
+    assert owner_email_set == mailoutbox_to_email_set
+
+
+@pytest.mark.django_db()
+def test_unembargo_dandiset_task_failure(draft_version_factory, mailoutbox):
+    # Intentionally set the status to embargoed so the task will fail
+    draft_version = draft_version_factory(dandiset__embargo_status=Dandiset.EmbargoStatus.EMBARGOED)
+    ds: Dandiset = draft_version.dandiset
+
+    with pytest.raises(DandiError):
+        unembargo_dandiset_task.delay(ds.pk)
+
+    assert mailoutbox
+    assert 'Unembargo failed' in mailoutbox[0].subject
+    payload = mailoutbox[0].message().get_payload()[0].get_payload()
+    assert ds.identifier in payload
+    assert 'error during the unembargo' in payload
diff --git a/dandiapi/api/tests/test_upload.py b/dandiapi/api/tests/test_upload.py
index 133c2ff7a..84c1b7bb2 100644
--- a/dandiapi/api/tests/test_upload.py
+++ b/dandiapi/api/tests/test_upload.py
@@ -110,7 +110,7 @@ def test_upload_initialize(api_client, user, dandiset_factory, embargoed):
 
 
 @pytest.mark.django_db()
-def test_upload_initialize_unembargoing(api_client, user, dandiset_factory):
+def test_upload_initialize_unembargo_in_progress(api_client, user, dandiset_factory):
     dandiset = dandiset_factory(embargo_status=Dandiset.EmbargoStatus.UNEMBARGOING)
     api_client.force_authenticate(user=user)
     assign_perm('owner', user, dandiset)
diff --git a/dandiapi/api/tests/test_version.py b/dandiapi/api/tests/test_version.py
index f25ae29cb..08b615ebc 100644
--- a/dandiapi/api/tests/test_version.py
+++ b/dandiapi/api/tests/test_version.py
@@ -606,7 +606,7 @@ def test_version_rest_update(api_client, user, draft_version):
 
 
 @pytest.mark.django_db()
-def test_version_rest_update_unembargoing(api_client, user, draft_version_factory):
+def test_version_rest_update_unembargo_in_progress(api_client, user, draft_version_factory):
     draft_version = draft_version_factory(
         dandiset__embargo_status=Dandiset.EmbargoStatus.UNEMBARGOING
     )
@@ -802,7 +802,7 @@ def test_version_rest_publish_embargo(api_client: APIClient, user: User, draft_v
 
 
 @pytest.mark.django_db()
-def test_version_rest_publish_unembargoing(
+def test_version_rest_publish_unembargo_in_progress(
     api_client: APIClient, user: User, draft_version_factory
 ):
     draft_version = draft_version_factory(
diff --git a/dandiapi/api/views/asset.py b/dandiapi/api/views/asset.py
index d92fa0d0e..5a3502e53 100644
--- a/dandiapi/api/views/asset.py
+++ b/dandiapi/api/views/asset.py
@@ -324,7 +324,7 @@ def create(self, request, versions__dandiset__pk, versions__version):
             version=versions__version,
         )
 
-        if version.dandiset.embargo_status == Dandiset.EmbargoStatus.UNEMBARGOING:
+        if version.dandiset.unembargo_in_progress:
             raise DandisetUnembargoInProgressError
 
         serializer = AssetRequestSerializer(data=self.request.data)
@@ -361,7 +361,7 @@ def update(self, request, versions__dandiset__pk, versions__version, **kwargs):
         )
         if version.version != 'draft':
             raise DraftDandisetNotModifiableError
-        if version.dandiset.embargo_status == Dandiset.EmbargoStatus.UNEMBARGOING:
+        if version.dandiset.unembargo_in_progress:
             raise DandisetUnembargoInProgressError
 
         serializer = AssetRequestSerializer(data=self.request.data)
@@ -399,7 +399,7 @@ def destroy(self, request, versions__dandiset__pk, versions__version, **kwargs):
             dandiset__pk=versions__dandiset__pk,
             version=versions__version,
         )
-        if version.dandiset.embargo_status == Dandiset.EmbargoStatus.UNEMBARGOING:
+        if version.dandiset.unembargo_in_progress:
             raise DandisetUnembargoInProgressError
 
         # Lock asset for delete
diff --git a/dandiapi/api/views/dandiset.py b/dandiapi/api/views/dandiset.py
index 3f1f561b4..6d6f9fa48 100644
--- a/dandiapi/api/views/dandiset.py
+++ b/dandiapi/api/views/dandiset.py
@@ -26,7 +26,7 @@
 from dandiapi.api.mail import send_ownership_change_emails
 from dandiapi.api.models import AuditRecord, Dandiset, Version
 from dandiapi.api.services.dandiset import create_dandiset, delete_dandiset
-from dandiapi.api.services.embargo import unembargo_dandiset
+from dandiapi.api.services.embargo import kickoff_dandiset_unembargo
 from dandiapi.api.services.embargo.exceptions import (
     DandisetUnembargoInProgressError,
     UnauthorizedEmbargoAccessError,
@@ -350,7 +350,7 @@ def destroy(self, request, dandiset__pk):
     @method_decorator(permission_required_or_403('owner', (Dandiset, 'pk', 'dandiset__pk')))
     def unembargo(self, request, dandiset__pk):
         dandiset: Dandiset = get_object_or_404(Dandiset, pk=dandiset__pk)
-        unembargo_dandiset(user=request.user, dandiset=dandiset)
+        kickoff_dandiset_unembargo(user=request.user, dandiset=dandiset)
 
         return Response(None, status=status.HTTP_200_OK)
 
@@ -378,7 +378,7 @@ def unembargo(self, request, dandiset__pk):
     def users(self, request, dandiset__pk):  # noqa: C901
         dandiset: Dandiset = self.get_object()
         if request.method == 'PUT':
-            if dandiset.embargo_status == Dandiset.EmbargoStatus.UNEMBARGOING:
+            if dandiset.unembargo_in_progress:
                 raise DandisetUnembargoInProgressError
 
             # Verify that the user is currently an owner
diff --git a/dandiapi/api/views/upload.py b/dandiapi/api/views/upload.py
index acdf6a41e..433195716 100644
--- a/dandiapi/api/views/upload.py
+++ b/dandiapi/api/views/upload.py
@@ -137,7 +137,7 @@ def upload_initialize_view(request: Request) -> HttpResponseBase:
         return response
 
     # Ensure dandiset not in the process of unembargo
-    if dandiset.embargo_status == Dandiset.EmbargoStatus.UNEMBARGOING:
+    if dandiset.unembargo_in_progress:
         raise DandisetUnembargoInProgressError
 
     logging.info(
diff --git a/dandiapi/api/views/version.py b/dandiapi/api/views/version.py
index 26c2c56ce..dffec235b 100644
--- a/dandiapi/api/views/version.py
+++ b/dandiapi/api/views/version.py
@@ -95,7 +95,7 @@ def update(self, request, **kwargs):
                 'Only draft versions can be modified.',
                 status=status.HTTP_405_METHOD_NOT_ALLOWED,
             )
-        if version.dandiset.embargo_status == Dandiset.EmbargoStatus.UNEMBARGOING:
+        if version.dandiset.unembargo_in_progress:
             raise DandisetUnembargoInProgressError
 
         serializer = VersionMetadataSerializer(data=request.data)
diff --git a/web/src/views/DandisetLandingView/DandisetActions.vue b/web/src/views/DandisetLandingView/DandisetActions.vue
index fc9ae4391..9249090ee 100644
--- a/web/src/views/DandisetLandingView/DandisetActions.vue
+++ b/web/src/views/DandisetLandingView/DandisetActions.vue
@@ -72,7 +72,7 @@
           id="view-data"
           outlined
           block
-          :disabled="currentDandiset.dandiset.embargo_status === 'UNEMBARGOING'"
+          :disabled="unembargo_in_progress"
           :to="fileBrowserLink"
           exact
         >
@@ -156,6 +156,7 @@ const store = useDandisetStore();
 
 const currentDandiset = computed(() => store.dandiset);
 const currentVersion = computed(() => store.version);
+const unembargo_in_progress = computed(() => currentDandiset.value && currentDandiset.value.dandiset.embargo_status === 'UNEMBARGOING')
 
 const fileBrowserLink: ComputedRef<Location|null> = computed(() => {
   if (!currentDandiset.value) {
diff --git a/web/src/views/DandisetLandingView/DandisetUnembargo.vue b/web/src/views/DandisetLandingView/DandisetUnembargo.vue
index c6c846a80..e1fbfc520 100644
--- a/web/src/views/DandisetLandingView/DandisetUnembargo.vue
+++ b/web/src/views/DandisetLandingView/DandisetUnembargo.vue
@@ -65,7 +65,7 @@
     >
       <v-tooltip
         left
-        :disabled="!unembargoing"
+        :disabled="!unembargo_in_progress"
       >
         <template #activator="{ on }">
           <div
@@ -77,16 +77,16 @@
               block
               color="info"
               depressed
-              :disabled="unembargoing"
+              :disabled="unembargo_in_progress"
               @click="unembargo()"
             >
-              {{ unembargoing ? 'Unembargoing' : 'Unembargo' }}
+              {{ unembargo_in_progress ? 'Unembargoing' : 'Unembargo' }}
               <v-spacer />
               <v-icon>mdi-lock-open</v-icon>
             </v-btn>
           </div>
         </template>
-        <span v-if="unembargoing">This dandiset is being unembargoed, please wait.</span>
+        <span v-if="unembargo_in_progress">This dandiset is being unembargoed, please wait.</span>
       </v-tooltip>
     </v-row>
     <v-row>
@@ -126,7 +126,7 @@ function formatDate(date: string): string {
 const store = useDandisetStore();
 
 const currentDandiset = computed(() => store.dandiset);
-const unembargoing = computed(() => currentDandiset.value?.dandiset.embargo_status === 'UNEMBARGOING');
+const unembargo_in_progress = computed(() => currentDandiset.value?.dandiset.embargo_status === 'UNEMBARGOING');
 const showWarningDialog = ref(false);
 const confirmationPhrase = ref('');
 
diff --git a/web/src/views/FileBrowserView/FileBrowser.vue b/web/src/views/FileBrowserView/FileBrowser.vue
index 0a7c6a89e..55f19a143 100644
--- a/web/src/views/FileBrowserView/FileBrowser.vue
+++ b/web/src/views/FileBrowserView/FileBrowser.vue
@@ -1,5 +1,5 @@
 <template>
-  <div v-if="currentDandiset && currentDandiset.dandiset.embargo_status !== 'UNEMBARGOING'">
+  <div v-if="!unembargo_in_progress">
     <v-progress-linear
       v-if="!currentDandiset"
       indeterminate
@@ -390,6 +390,7 @@ const updating = ref(false);
 const owners = computed(() => store.owners?.map((u) => u.username) || null);
 const currentDandiset = computed(() => store.dandiset);
 const embargoed = computed(() => currentDandiset.value?.dandiset.embargo_status === 'EMBARGOED');
+const unembargo_in_progress = computed(() => currentDandiset.value?.dandiset.embargo_status === 'UNEMBARGOING')
 const splitLocation = computed(() => location.value.split('/'));
 const isAdmin = computed(() => user.value?.admin || false);
 const isOwner = computed(() => !!(