diff --git a/static/js/settings_exports.js b/static/js/settings_exports.js
index 42d7213197..79e687bd70 100644
--- a/static/js/settings_exports.js
+++ b/static/js/settings_exports.js
@@ -38,6 +38,8 @@ exports.populate_exports_table = function (exports) {
                             new XDate(data.export_time * 1000)
                         ),
                         url: data.export_url,
+                        failed: data.failed_timestamp,
+                        pending: data.pending,
                     },
                 });
             }
diff --git a/static/templates/admin_export_list.hbs b/static/templates/admin_export_list.hbs
index 4986589bd6..d6baf45bc0 100644
--- a/static/templates/admin_export_list.hbs
+++ b/static/templates/admin_export_list.hbs
@@ -6,11 +6,18 @@
     <td>
         <span class="export_time">{{event_time}}</span>
     </td>
+    <td>
+        {{#if url}}
+        <span class="export_status">{{t 'Complete' }}</span>
+        {{else if failed}}
+        <span class="export_status">{{t 'Failed' }}</span>
+        {{else if pending}}
+        <span class="export_status">{{t 'Pending' }}</span>
+        {{/if}}
+    </td>
     <td>
         {{#if url}}
         <span class="export_url"><a href="{{url}}" download>{{t 'Download' }}</a></span>
-        {{else}}
-        <span class="export_url">{{t 'The export URL is not yet available... Check back soon.' }}</span>
         {{/if}}
     </td>
     <td class="actions">
diff --git a/static/templates/settings/data_exports_admin.hbs b/static/templates/settings/data_exports_admin.hbs
index 110a2478d2..64e88a42a8 100644
--- a/static/templates/settings/data_exports_admin.hbs
+++ b/static/templates/settings/data_exports_admin.hbs
@@ -35,6 +35,7 @@
             <thead>
                 <th class="active" data-sort="user">{{t "Requesting user" }}</th>
                 <th data-sort="numeric" data-sort-prop="export_time">{{t "Time" }}</th>
+                <th>{{t "Status" }}</th>
                 <th>{{t "File" }}</th>
                 <th class="actions">{{t "Actions" }}</th>
             </thead>
diff --git a/zerver/lib/actions.py b/zerver/lib/actions.py
index 846e4333d2..d31a8b984c 100644
--- a/zerver/lib/actions.py
+++ b/zerver/lib/actions.py
@@ -5798,8 +5798,12 @@ def do_delete_realm_export(user_profile: UserProfile, export: RealmAuditLog) ->
     export_extra_data = export.extra_data
     assert export_extra_data is not None
     export_data = ujson.loads(export_extra_data)
+    export_path = export_data.get('export_path')
+
+    if export_path:
+        # Allow removal even if the export failed.
+        delete_export_tarball(export_path)
 
-    delete_export_tarball(export_data.get('export_path'))
     export_data.update({'deleted_timestamp': timezone_now().timestamp()})
     export.extra_data = ujson.dumps(export_data)
     export.save(update_fields=['extra_data'])
diff --git a/zerver/lib/export.py b/zerver/lib/export.py
index 60e037c740..90908924bf 100644
--- a/zerver/lib/export.py
+++ b/zerver/lib/export.py
@@ -1763,14 +1763,30 @@ def get_realm_exports_serialized(user: UserProfile) -> List[Dict[str, Any]]:
                                                event_type=RealmAuditLog.REALM_EXPORTED)
     exports_dict = {}
     for export in all_exports:
-        export_data = ujson.loads(export.extra_data)
-        export_url = zerver.lib.upload.upload_backend.get_export_tarball_url(
-            user.realm, export_data['export_path'])
+        pending = True
+        export_url = None
+        deleted_timestamp = None
+        failed_timestamp = None
+
+        if export.extra_data is not None:
+            pending = False
+
+            export_data = ujson.loads(export.extra_data)
+            deleted_timestamp = export_data.get('deleted_timestamp')
+            failed_timestamp = export_data.get('failed_timestamp')
+            export_path = export_data.get('export_path')
+
+            if export_path:
+                export_url = zerver.lib.upload.upload_backend.get_export_tarball_url(
+                    user.realm, export_path)
+
         exports_dict[export.id] = dict(
             id=export.id,
             export_time=export.event_time.timestamp(),
             acting_user_id=export.acting_user.id,
             export_url=export_url,
-            deleted_timestamp=export_data.get('deleted_timestamp'),
+            deleted_timestamp=deleted_timestamp,
+            failed_timestamp=failed_timestamp,
+            pending=pending,
         )
     return sorted(exports_dict.values(), key=lambda export_dict: export_dict['id'])
diff --git a/zerver/tests/test_events.py b/zerver/tests/test_events.py
index f059616213..bca1e3b527 100644
--- a/zerver/tests/test_events.py
+++ b/zerver/tests/test_events.py
@@ -2813,6 +2813,19 @@ class EventsRegisterTest(ZulipTestCase):
         self.assert_on_error(error)
 
     def test_notify_realm_export(self) -> None:
+        pending_schema_checker = self.check_events_dict([
+            ('type', equals('realm_export')),
+            ('exports', check_list(check_dict_only([
+                ('id', check_int),
+                ('export_time', check_float),
+                ('acting_user_id', check_int),
+                ('export_url', equals(None)),
+                ('deleted_timestamp', equals(None)),
+                ('failed_timestamp', equals(None)),
+                ('pending', check_bool),
+            ]))),
+        ])
+
         schema_checker = self.check_events_dict([
             ('type', equals('realm_export')),
             ('exports', check_list(check_dict_only([
@@ -2821,6 +2834,8 @@ class EventsRegisterTest(ZulipTestCase):
                 ('acting_user_id', check_int),
                 ('export_url', check_string),
                 ('deleted_timestamp', equals(None)),
+                ('failed_timestamp', equals(None)),
+                ('pending', check_bool),
             ]))),
         ])
 
@@ -2832,10 +2847,14 @@ class EventsRegisterTest(ZulipTestCase):
             with stdout_suppressed():
                 events = self.do_test(
                     lambda: self.client_post('/json/export/realm'),
-                    state_change_expected=True, num_events=2)
+                    state_change_expected=True, num_events=3)
 
-        # The first event is a message from notification-bot.
-        error = schema_checker('events[1]', events[1])
+        # We first notify when an export is initiated,
+        error = pending_schema_checker('events[0]', events[0])
+        self.assert_on_error(error)
+
+        # The second event is then a message from notification-bot.
+        error = schema_checker('events[2]', events[2])
         self.assert_on_error(error)
 
         # Now we check the deletion of the export.
@@ -2847,6 +2866,8 @@ class EventsRegisterTest(ZulipTestCase):
                 ('acting_user_id', check_int),
                 ('export_url', check_string),
                 ('deleted_timestamp', check_float),
+                ('failed_timestamp', equals(None)),
+                ('pending', check_bool),
             ]))),
         ])
 
@@ -2858,6 +2879,49 @@ class EventsRegisterTest(ZulipTestCase):
         error = deletion_schema_checker('events[0]', events[0])
         self.assert_on_error(error)
 
+    def test_notify_realm_export_on_failure(self) -> None:
+        pending_schema_checker = self.check_events_dict([
+            ('type', equals('realm_export')),
+            ('exports', check_list(check_dict_only([
+                ('id', check_int),
+                ('export_time', check_float),
+                ('acting_user_id', check_int),
+                ('export_url', equals(None)),
+                ('deleted_timestamp', equals(None)),
+                ('failed_timestamp', equals(None)),
+                ('pending', check_bool),
+            ]))),
+        ])
+
+        failed_schema_checker = self.check_events_dict([
+            ('type', equals('realm_export')),
+            ('exports', check_list(check_dict_only([
+                ('id', check_int),
+                ('export_time', check_float),
+                ('acting_user_id', check_int),
+                ('export_url', equals(None)),
+                ('deleted_timestamp', equals(None)),
+                ('failed_timestamp', check_float),
+                ('pending', check_bool),
+            ]))),
+        ])
+
+        do_change_is_admin(self.user_profile, True)
+        self.login_user(self.user_profile)
+
+        with mock.patch('zerver.lib.export.do_export_realm',
+                        side_effect=Exception("test")):
+            with stdout_suppressed():
+                events = self.do_test(
+                    lambda: self.client_post('/json/export/realm'),
+                    state_change_expected=False, num_events=2)
+
+        error = pending_schema_checker('events[0]', events[0])
+        self.assert_on_error(error)
+
+        error = failed_schema_checker('events[1]', events[1])
+        self.assert_on_error(error)
+
 class FetchInitialStateDataTest(ZulipTestCase):
     # Non-admin users don't have access to all bots
     def test_realm_bots_non_admin(self) -> None:
diff --git a/zerver/views/realm_export.py b/zerver/views/realm_export.py
index 5f3f67e18a..4b7b777fc8 100644
--- a/zerver/views/realm_export.py
+++ b/zerver/views/realm_export.py
@@ -12,7 +12,7 @@ from zerver.models import RealmAuditLog, UserProfile
 from zerver.lib.queue import queue_json_publish
 from zerver.lib.response import json_error, json_success
 from zerver.lib.export import get_realm_exports_serialized
-from zerver.lib.actions import do_delete_realm_export
+from zerver.lib.actions import do_delete_realm_export, notify_realm_export
 
 import ujson
 
@@ -51,6 +51,10 @@ def export_realm(request: HttpRequest, user: UserProfile) -> HttpResponse:
                                        event_type=event_type,
                                        event_time=event_time,
                                        acting_user=user)
+
+    # Allow for UI updates on a pending export
+    notify_realm_export(user)
+
     # Using the deferred_work queue processor to avoid
     # killing the process after 60s
     event = {'type': "realm_export",
diff --git a/zerver/worker/queue_processors.py b/zerver/worker/queue_processors.py
index a8ad186f67..0022c2d381 100644
--- a/zerver/worker/queue_processors.py
+++ b/zerver/worker/queue_processors.py
@@ -13,6 +13,7 @@ import socket
 
 from django.conf import settings
 from django.db import connection
+from django.utils.timezone import now as timezone_now
 from zerver.models import \
     get_client, get_system_bot, PreregistrationUser, \
     get_user_profile_by_id, Message, Realm, UserMessage, UserProfile, \
@@ -680,14 +681,26 @@ class DeferredWorker(QueueProcessingWorker):
             start = time.time()
             realm = Realm.objects.get(id=event['realm_id'])
             output_dir = tempfile.mkdtemp(prefix="zulip-export-")
+            export_event = RealmAuditLog.objects.get(id=event['id'])
+            user_profile = get_user_profile_by_id(event['user_profile_id'])
+
+            try:
+                public_url = export_realm_wrapper(realm=realm, output_dir=output_dir,
+                                                  threads=6, upload=True, public_only=True,
+                                                  delete_after_upload=True)
+            except Exception:
+                export_event.extra_data = ujson.dumps(dict(
+                    failed_timestamp=timezone_now().timestamp()
+                ))
+                export_event.save(update_fields=['extra_data'])
+                logging.error("Data export for %s failed after %s" % (
+                    user_profile.realm.string_id, time.time() - start))
+                notify_realm_export(user_profile)
+                return
 
-            public_url = export_realm_wrapper(realm=realm, output_dir=output_dir,
-                                              threads=6, upload=True, public_only=True,
-                                              delete_after_upload=True)
             assert public_url is not None
 
             # Update the extra_data field now that the export is complete.
-            export_event = RealmAuditLog.objects.get(id=event['id'])
             export_event.extra_data = ujson.dumps(dict(
                 export_path=urllib.parse.urlparse(public_url).path,
             ))
@@ -695,7 +708,6 @@ class DeferredWorker(QueueProcessingWorker):
 
             # Send a private message notification letting the user who
             # triggered the export know the export finished.
-            user_profile = get_user_profile_by_id(event['user_profile_id'])
             content = "Your data export is complete and has been uploaded here:\n\n%s" % (
                 public_url,)
             internal_send_private_message(