diff --git a/fiesta/apps/notifications/management/commands/send_scheduled_notifications.py b/fiesta/apps/notifications/management/commands/send_scheduled_notifications.py
index e5747279..0427bfda 100644
--- a/fiesta/apps/notifications/management/commands/send_scheduled_notifications.py
+++ b/fiesta/apps/notifications/management/commands/send_scheduled_notifications.py
@@ -26,19 +26,57 @@
 class Command(BaseCommand):
     help = "Send all pending scheduled notifications whose send_after time has passed."
 
+    def add_arguments(self, parser: object) -> None:
+        parser.add_argument(
+            "--batch-size",
+            type=int,
+            default=100,
+            help="Maximum number of pending notifications to process in a single run.",
+        )
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            help="Log pending notifications without sending or updating sent_at.",
+        )
+
     def handle(self, *args: object, **options: object) -> None:
+        batch_size = options["batch_size"]
+        dry_run = options["dry_run"]
         now = timezone.now()
 
+        # Dry-run: just log pending notifications without claiming them.
+        if dry_run:
+            pending_qs = ScheduledNotification.objects.filter(
+                send_after__lte=now,
+                sent_at__isnull=True,
+                cancelled_at__isnull=True,
+            ).select_related("recipient")[:batch_size]
+            would_send = 0
+            for notification in pending_qs:
+                logger.info(
+                    "Dry-run: would send scheduled notification pk=%s kind=%s",
+                    notification.pk,
+                    notification.kind,
+                )
+                would_send += 1
+            self.stdout.write(self.style.SUCCESS(f"Dry-run complete: would send {would_send} notification(s)."))
+            return
+
         # Phase 1: Atomically claim pending notifications by stamping sent_at.
         # select_for_update(skip_locked=True) + immediate UPDATE prevents concurrent
         # workers from picking the same rows (locks held only for the UPDATE, not I/O).
         with transaction.atomic():
-            pending_qs = ScheduledNotification.objects.select_for_update(skip_locked=True).filter(
-                send_after__lte=now,
-                sent_at__isnull=True,
-                cancelled_at__isnull=True,
+            pending_qs = (
+                ScheduledNotification.objects.select_for_update(skip_locked=True)
+                .filter(
+                    send_after__lte=now,
+                    sent_at__isnull=True,
+                    cancelled_at__isnull=True,
+                )
+                .order_by("send_after", "pk")
             )
-            reserved_ids: list[int] = list(pending_qs.values_list("pk", flat=True))
+
+            reserved_ids: list[int] = list(pending_qs.values_list("pk", flat=True)[:batch_size])
             if reserved_ids:
                 ScheduledNotification.objects.filter(pk__in=reserved_ids).update(sent_at=now)
 
@@ -46,9 +84,12 @@ def handle(self, *args: object, **options: object) -> None:
             self.stdout.write("No pending notifications.")
             return
 
+        logger.info("Claimed %d notification(s) for sending (batch size: %d)", len(reserved_ids), batch_size)
+
         # Phase 2: Send each notification outside any transaction (no DB locks during I/O).
         sent = 0
         skipped = 0
+        failed = 0
 
         for notification_pk in reserved_ids:
             try:
@@ -76,9 +117,9 @@ def handle(self, *args: object, **options: object) -> None:
                 logger.exception("Failed to send scheduled notification pk=%s", notification.pk)
                 # Roll back the claim so the notification can be retried next run.
                 ScheduledNotification.objects.filter(pk=notification_pk).update(sent_at=None)
-                skipped += 1
+                failed += 1
 
-        self.stdout.write(self.style.SUCCESS(f"Sent {sent} notification(s), skipped {skipped}."))
+        self.stdout.write(self.style.SUCCESS(f"Sent {sent} notification(s), skipped {skipped}, failed {failed}."))
 
     def _send(self, notification: ScheduledNotification) -> bool:
         # Resolve the related object via GenericFK
diff --git a/fiesta/apps/notifications/migrations/0002_alter_sectionnotificationpreferences_options_and_more.py b/fiesta/apps/notifications/migrations/0002_alter_sectionnotificationpreferences_options_and_more.py
new file mode 100644
index 00000000..579d6282
--- /dev/null
+++ b/fiesta/apps/notifications/migrations/0002_alter_sectionnotificationpreferences_options_and_more.py
@@ -0,0 +1,35 @@
+# Generated by Django 4.2.28 on 2026-03-07 07:49
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('contenttypes', '0002_remove_content_type_name'),
+        ('sections', '0024_initial'),
+        ('notifications', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name='sectionnotificationpreferences',
+            options={'verbose_name': 'section notification preference', 'verbose_name_plural': 'section notification preferences'},
+        ),
+        migrations.AlterField(
+            model_name='schedulednotification',
+            name='content_type',
+            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to='contenttypes.contenttype', verbose_name='content type'),
+        ),
+        migrations.AlterField(
+            model_name='schedulednotification',
+            name='kind',
+            field=models.CharField(choices=[('buddy_matched_issuer', 'Buddy matched — issuer'), ('pickup_matched_issuer', 'Pickup matched — issuer'), ('member_waiting_digest', 'New member waiting digest')], max_length=64, verbose_name='kind'),
+        ),
+        migrations.AlterField(
+            model_name='schedulednotification',
+            name='section',
+            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='scheduled_notifications', to='sections.section', verbose_name='section'),
+        ),
+    ]
diff --git a/fiesta/apps/notifications/models/scheduled.py b/fiesta/apps/notifications/models/scheduled.py
index f3a9938b..6813ac2f 100644
--- a/fiesta/apps/notifications/models/scheduled.py
+++ b/fiesta/apps/notifications/models/scheduled.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from django.conf import settings
 from django.contrib.contenttypes.fields import GenericForeignKey
 from django.contrib.contenttypes.models import ContentType
 from django.db import models
@@ -32,7 +33,7 @@ class ScheduledNotification(BaseTimestampedModel):
     )
 
     recipient = models.ForeignKey(
-        "accounts.User",
+        settings.AUTH_USER_MODEL,
         on_delete=models.CASCADE,
         related_name="scheduled_notifications",
         verbose_name=_("recipient"),
@@ -40,6 +41,8 @@ class ScheduledNotification(BaseTimestampedModel):
     section = models.ForeignKey(
         "sections.Section",
         on_delete=models.CASCADE,
+        null=True,
+        blank=True,
         related_name="scheduled_notifications",
         verbose_name=_("section"),
     )
diff --git a/fiesta/apps/notifications/services/mailer.py b/fiesta/apps/notifications/services/mailer.py
index 17e598f5..47f8139a 100644
--- a/fiesta/apps/notifications/services/mailer.py
+++ b/fiesta/apps/notifications/services/mailer.py
@@ -9,6 +9,8 @@
 from django.core.mail import EmailMultiAlternatives
 from django.template.loader import render_to_string
 
+from apps.notifications.services.unsubscribe import generate_unsubscribe_token
+
 logger = logging.getLogger(__name__)
 
 if typing.TYPE_CHECKING:
@@ -37,19 +39,31 @@ def send_notification_email(
         except ObjectDoesNotExist:
             pass  # No profile — proceed with sending
 
-    html_content = render_to_string(f"{template_prefix}.html", context)
-    text_content = render_to_string(f"{template_prefix}.txt", context)
+    email_context = dict(context)
+    unsubscribe_url = ""
+    if recipient_user is not None:
+        token = generate_unsubscribe_token(recipient_user.pk)
+        unsubscribe_url = f"https://{settings.ROOT_DOMAIN}/notifications/unsubscribe/{token}/"
+
+    email_context["unsubscribe_url"] = unsubscribe_url
 
-    msg = EmailMultiAlternatives(
+    html_content = render_to_string(f"{template_prefix}.html", email_context)
+    text_content = render_to_string(f"{template_prefix}.txt", email_context)
+
+    email = EmailMultiAlternatives(
         subject=subject,
         body=text_content,
         from_email=settings.DEFAULT_FROM_EMAIL,
         to=[recipient_email],
     )
-    msg.attach_alternative(html_content, "text/html")
+    if unsubscribe_url:
+        email.extra_headers["List-Unsubscribe"] = f"<{unsubscribe_url}>"
+        email.extra_headers["List-Unsubscribe-Post"] = "List-Unsubscribe=One-Click"
+
+    email.attach_alternative(html_content, "text/html")
 
     try:
-        msg.send()
+        email.send()
     except Exception:
         logger.exception(
             "Failed to send notification email (template: %s)",
diff --git a/fiesta/apps/notifications/services/scheduler.py b/fiesta/apps/notifications/services/scheduler.py
index 8cd438bf..bc3de1a0 100644
--- a/fiesta/apps/notifications/services/scheduler.py
+++ b/fiesta/apps/notifications/services/scheduler.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+from contextlib import suppress
 from datetime import datetime
 from typing import TYPE_CHECKING
 
@@ -36,9 +37,10 @@ def enqueue_delayed_notification(
 
     Uses select_for_update to prevent duplicate rows from concurrent requests.
     """
-    if hasattr(recipient, "profile") and not recipient.profile.email_notifications_enabled:
-        logger.info("Skipping scheduled notification for %s: global opt-out", recipient)
-        return None
+    with suppress(AttributeError):
+        if not recipient.profile.email_notifications_enabled:
+            logger.info("Skipping scheduled notification for %s: global opt-out", recipient)
+            return None
 
     ct = ContentType.objects.get_for_model(related_object)
 
diff --git a/fiesta/apps/notifications/services/unsubscribe.py b/fiesta/apps/notifications/services/unsubscribe.py
new file mode 100644
index 00000000..1352518f
--- /dev/null
+++ b/fiesta/apps/notifications/services/unsubscribe.py
@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+from django.core.signing import BadSignature, TimestampSigner
+
+UNSUBSCRIBE_SALT = "notifications-unsubscribe"
+
+
+def generate_unsubscribe_token(user_id: int | str, action: str = "global") -> str:
+    signer = TimestampSigner(salt=UNSUBSCRIBE_SALT)
+    return signer.sign(f"{user_id}:{action}")
+
+
+def verify_unsubscribe_token(token: str, max_age: int = 30 * 24 * 60 * 60) -> tuple[str, str]:
+    signer = TimestampSigner(salt=UNSUBSCRIBE_SALT)
+    value = signer.unsign(token, max_age=max_age)
+    try:
+        user_id_str, action = value.rsplit(":", 1)
+        return user_id_str, action
+    except (TypeError, ValueError) as exc:
+        raise BadSignature("Invalid unsubscribe payload") from exc
diff --git a/fiesta/apps/notifications/templates/notifications/base_email.html b/fiesta/apps/notifications/templates/notifications/base_email.html
index 711f1391..cf9aeead 100644
--- a/fiesta/apps/notifications/templates/notifications/base_email.html
+++ b/fiesta/apps/notifications/templates/notifications/base_email.html
@@ -38,6 +38,12 @@ <h1>
         <p>
           {% if section %}You are receiving this email because you are a member of {{ section }}.{% endif %}
           {% if preferences_url %}<a href="{{ preferences_url }}">Manage email preferences</a>{% endif %}
+          {% if unsubscribe_url %}
+            |
+            <a href="{{ unsubscribe_url }}"
+               style="color: #6b7280;
+                      text-decoration: underline">Unsubscribe</a>
+          {% endif %}
         </p>
         <p>
           © Fiesta+ · <a href="https://fiesta.plus">fiesta.plus</a>
diff --git a/fiesta/apps/notifications/templates/notifications/base_email.txt b/fiesta/apps/notifications/templates/notifications/base_email.txt
index ec8440ef..84b755f5 100644
--- a/fiesta/apps/notifications/templates/notifications/base_email.txt
+++ b/fiesta/apps/notifications/templates/notifications/base_email.txt
@@ -7,3 +7,4 @@ Fiesta+{% if section %} · {{ section }}{% endif %}
 You are receiving this email because you are a member of {{ section }}.
 {% if preferences_url %}Manage preferences: {{ preferences_url }}{% endif %}
 © Fiesta+ · https://fiesta.plus
+{% if unsubscribe_url %}Unsubscribe: {{ unsubscribe_url }}{% endif %}
diff --git a/fiesta/apps/notifications/templates/notifications/unsubscribe.html b/fiesta/apps/notifications/templates/notifications/unsubscribe.html
new file mode 100644
index 00000000..8263ee0c
--- /dev/null
+++ b/fiesta/apps/notifications/templates/notifications/unsubscribe.html
@@ -0,0 +1,26 @@
+{% extends "fiesta/base.html" %}
+{% load i18n %}
+
+{% block main %}
+  <div class="container mx-auto max-w-lg py-8">
+    {% if error %}
+      <h1 class="text-2xl font-bold mb-4">{% trans "Unsubscribe Error" %}</h1>
+      <p>{{ error }}</p>
+    {% elif success %}
+      <h1 class="text-2xl font-bold mb-4">{% trans "Unsubscribed" %}</h1>
+      <p>
+        {% blocktrans with action=action_description %}You have been successfully unsubscribed from {{ action }}.{% endblocktrans %}
+      </p>
+    {% else %}
+      <h1 class="text-2xl font-bold mb-4">{% trans "Unsubscribe from Notifications" %}</h1>
+      <p>
+        {% blocktrans with email=email action=action_description %}
+                You are about to unsubscribe {{ email }} from {{ action }}.
+            {% endblocktrans %}
+      </p>
+      <form method="post">
+        <button type="submit" class="btn btn-primary mt-4">{% trans "Unsubscribe" %}</button>
+      </form>
+    {% endif %}
+  </div>
+{% endblock main %}
diff --git a/fiesta/apps/notifications/tests/factories.py b/fiesta/apps/notifications/tests/factories.py
index e8a5ab75..074d0f3d 100644
--- a/fiesta/apps/notifications/tests/factories.py
+++ b/fiesta/apps/notifications/tests/factories.py
@@ -7,7 +7,7 @@
 from django.utils import timezone
 from factory.django import DjangoModelFactory
 
-from apps.notifications.models import ScheduledNotification, SectionNotificationPreferences
+from apps.notifications.models import NotificationKind, ScheduledNotification, SectionNotificationPreferences
 
 
 class ScheduledNotificationFactory(DjangoModelFactory):
@@ -16,8 +16,8 @@ class ScheduledNotificationFactory(DjangoModelFactory):
     class Meta:
         model = ScheduledNotification
 
-    kind = ScheduledNotification.Kind.BUDDY_MATCHED_ISSUER
-    recipient = factory.SubFactory("apps.utils.factories.accounts.UserFactory")
+    kind = NotificationKind.BUDDY_MATCHED_ISSUER
+    recipient = factory.SubFactory("apps.utils.factories.accounts.UserFactory", profile=None)
     section = factory.SubFactory("apps.utils.factories.sections.KnownSectionFactory")
 
     # Generic FK fields — point to a section by default (simple object that always exists)
diff --git a/fiesta/apps/notifications/tests/test_match_notifications.py b/fiesta/apps/notifications/tests/test_match_notifications.py
index feb07a1d..b813a14e 100644
--- a/fiesta/apps/notifications/tests/test_match_notifications.py
+++ b/fiesta/apps/notifications/tests/test_match_notifications.py
@@ -6,13 +6,19 @@
 
 from django.test import TestCase
 
-from apps.notifications.models import ScheduledNotification
+from apps.accounts.models import UserProfile
+from apps.notifications.models import NotificationKind, ScheduledNotification
 from apps.notifications.services.match import notify_buddy_match, notify_pickup_match
 from apps.notifications.tests.factories import SectionNotificationPreferencesFactory
 from apps.utils.factories.accounts import UserFactory
 from apps.utils.factories.sections import KnownSectionFactory
 
 
+def _create_user_profile(user):
+    """Ensure user has a UserProfile attached."""
+    UserProfile.objects.get_or_create(user=user)
+
+
 def _make_buddy_config(notify=True, delay=timedelta(hours=1)):
     """Return a mock BuddySystemConfiguration with the fields used by the service."""
     config = MagicMock()
@@ -59,7 +65,9 @@ class NotifyBuddyMatchTestCase(TestCase):
     def setUp(self):
         self.base_section = KnownSectionFactory()
         self.matcher = UserFactory(profile=None)
+        _create_user_profile(self.matcher)
         self.issuer = UserFactory(profile=None)
+        _create_user_profile(self.issuer)
         self.match, self.request_obj = _make_match_and_request(self.matcher, self.issuer)
 
     @patch("apps.notifications.services.match.send_notification_email")
@@ -87,7 +95,7 @@ def test_buddy_match_enqueues_issuer_notification(self, mock_send, mock_enqueue)
 
         mock_enqueue.assert_called_once()
         call_kwargs = mock_enqueue.call_args.kwargs
-        self.assertEqual(call_kwargs["kind"], ScheduledNotification.Kind.BUDDY_MATCHED_ISSUER)
+        self.assertEqual(call_kwargs["kind"], NotificationKind.BUDDY_MATCHED_ISSUER)
         self.assertEqual(call_kwargs["recipient"], self.issuer)
         self.assertEqual(call_kwargs["related_object"], self.match)
 
@@ -156,7 +164,9 @@ class NotifyPickupMatchTestCase(TestCase):
     def setUp(self):
         self.base_section = KnownSectionFactory()
         self.matcher = UserFactory(profile=None)
+        _create_user_profile(self.matcher)
         self.issuer = UserFactory(profile=None)
+        _create_user_profile(self.issuer)
         self.match, self.request_obj = _make_match_and_request(self.matcher, self.issuer)
 
     @patch("apps.notifications.services.match.send_notification_email")
diff --git a/fiesta/apps/notifications/tests/test_membership_notifications.py b/fiesta/apps/notifications/tests/test_membership_notifications.py
index 1913583b..8fd5e1e6 100644
--- a/fiesta/apps/notifications/tests/test_membership_notifications.py
+++ b/fiesta/apps/notifications/tests/test_membership_notifications.py
@@ -5,7 +5,8 @@
 
 from django.test import TestCase
 
-from apps.notifications.models import ScheduledNotification
+from apps.accounts.models import UserProfile
+from apps.notifications.models import NotificationKind, ScheduledNotification
 from apps.notifications.services.membership import notify_new_membership
 from apps.notifications.tests.factories import SectionNotificationPreferencesFactory
 from apps.sections.models import SectionMembership
@@ -13,6 +14,11 @@
 from apps.utils.factories.sections import KnownSectionFactory, SectionMembershipWithUserFactory
 
 
+def _create_user_profile(user):
+    """Ensure user has a UserProfile attached."""
+    UserProfile.objects.get_or_create(user=user)
+
+
 def _make_sections_config(
     notify_member=True,
     notify_on_new_member=True,
@@ -30,6 +36,7 @@ class NotifyNewMembershipTestCase(TestCase):
     def setUp(self):
         self.section = KnownSectionFactory()
         self.applicant = UserFactory(profile=None)
+        _create_user_profile(self.applicant)
         self.membership = SectionMembershipWithUserFactory(
             section=self.section,
             user=self.applicant,
@@ -78,7 +85,7 @@ def test_enqueues_digest_for_editors(self, mock_send, mock_enqueue):
         # Both editor and admin should have enqueue called
         self.assertEqual(mock_enqueue.call_count, 2)
         for call in mock_enqueue.call_args_list:
-            self.assertEqual(call.kwargs["kind"], ScheduledNotification.Kind.MEMBER_WAITING_DIGEST)
+            self.assertEqual(call.kwargs["kind"], NotificationKind.MEMBER_WAITING_DIGEST)
             self.assertEqual(call.kwargs["related_object"], self.membership)
 
     @patch("apps.notifications.services.scheduler.enqueue_delayed_notification")
diff --git a/fiesta/apps/notifications/tests/test_scheduled_command.py b/fiesta/apps/notifications/tests/test_scheduled_command.py
index b1a381ed..7adcc3c1 100644
--- a/fiesta/apps/notifications/tests/test_scheduled_command.py
+++ b/fiesta/apps/notifications/tests/test_scheduled_command.py
@@ -9,7 +9,8 @@
 from django.test import TestCase
 from django.utils import timezone
 
-from apps.notifications.models import ScheduledNotification
+from apps.accounts.models import UserProfile
+from apps.notifications.models import NotificationKind, ScheduledNotification
 from apps.notifications.tests.factories import ScheduledNotificationFactory
 from apps.sections.models import Section
 from apps.utils.factories.accounts import UserFactory
@@ -18,16 +19,24 @@
 COMMAND_NAME = "send_scheduled_notifications"
 
 
+def _make_user_with_profile():
+    """Create a User with an attached UserProfile, avoiding the broken UserProfileFactory."""
+    user = UserFactory(profile=None)
+    UserProfile.objects.create(user=user)
+    return user
+
+
 class SendScheduledNotificationsTestCase(TestCase):
     def setUp(self):
         self.section: Section = KnownSectionFactory()
-        self.recipient = UserFactory(profile=None)
+        self.recipient = _make_user_with_profile()
 
-    def _make_pending(self, send_after=None):
+    def _make_pending(self, kind=NotificationKind.MEMBER_WAITING_DIGEST, send_after=None):
         """Create a pending notification whose send_after is in the past by default."""
         return ScheduledNotificationFactory(
             recipient=self.recipient,
             section=self.section,
+            kind=kind,
             send_after=send_after or (timezone.now() - timedelta(minutes=5)),
         )
 
@@ -101,7 +110,7 @@ def test_soft_cancels_on_deleted_object(self, mock_send):
         notification = ScheduledNotificationFactory(
             recipient=self.recipient,
             section=self.section,
-            kind=ScheduledNotification.Kind.BUDDY_MATCHED_ISSUER,
+            kind=NotificationKind.BUDDY_MATCHED_ISSUER,
             send_after=timezone.now() - timedelta(minutes=5),
         )
         # Overwrite content_type/object_id to point to a deleted object
@@ -117,22 +126,39 @@ def test_soft_cancels_on_deleted_object(self, mock_send):
         mock_send.assert_not_called()
 
     @patch("apps.notifications.management.commands.send_scheduled_notifications.send_notification_email")
-    def test_select_for_update_used(self, mock_send):
-        """
-        Verify that the queryset uses select_for_update to prevent double-processing.
-        We test this by inspecting the command source and confirming the queryset chain
-        calls select_for_update before filter.
-        """
-        import inspect
+    def test_select_for_update_called_with_skip_locked(self, mock_send):
+        """The command acquires row locks with skip_locked to support concurrent workers."""
+        self._make_pending()
 
-        from apps.notifications.management.commands.send_scheduled_notifications import Command
+        with patch(
+            "apps.notifications.management.commands.send_scheduled_notifications.ScheduledNotification.objects.select_for_update",
+            wraps=ScheduledNotification.objects.select_for_update,
+        ) as mock_select_for_update:
+            call_command(COMMAND_NAME)
 
-        source = inspect.getsource(Command.handle)
-        self.assertIn("select_for_update", source)
+        mock_select_for_update.assert_called_once_with(skip_locked=True)
+
+    @patch("apps.notifications.management.commands.send_scheduled_notifications.send_notification_email")
+    def test_dry_run_logs_without_sending_or_marking_sent(self, mock_send):
+        """Dry-run logs pending notifications but does not send or persist sent_at."""
+        notification = self._make_pending()
+
+        with self.assertLogs(
+            "apps.notifications.management.commands.send_scheduled_notifications",
+            level="INFO",
+        ) as captured_logs:
+            call_command(COMMAND_NAME, dry_run=True)
+
+        notification.refresh_from_db()
+
+        self.assertIsNone(notification.sent_at)
+        self.assertIsNone(notification.cancelled_at)
+        self.assertIn("Dry-run: would send scheduled notification", "\n".join(captured_logs.output))
+        mock_send.assert_not_called()
 
     @patch("apps.notifications.management.commands.send_scheduled_notifications.send_notification_email")
     def test_send_failure_does_not_mark_sent(self, mock_send):
-        """When send_notification_email raises, sent_at must remain None."""
+        """When send_notification_email raises, sent_at must remain None (claim is rolled back)."""
         mock_send.side_effect = RuntimeError("SMTP connection failed")
         notification = self._make_pending()
 
@@ -160,3 +186,62 @@ def test_partial_failure_sends_remaining(self, mock_send):
         failed_count = sum(1 for n in [n1, n2] if n.sent_at is None)
         self.assertEqual(sent_count, 1)
         self.assertEqual(failed_count, 1)
+
+    @patch("apps.notifications.management.commands.send_scheduled_notifications.send_notification_email")
+    def test_send_failure_logs_error_and_continues(self, mock_send):
+        """A send exception is logged and processing continues to the next notification."""
+        first_recipient = _make_user_with_profile()
+        second_recipient = _make_user_with_profile()
+        first = ScheduledNotificationFactory(
+            recipient=first_recipient,
+            section=self.section,
+            kind=NotificationKind.MEMBER_WAITING_DIGEST,
+            send_after=timezone.now() - timedelta(minutes=5),
+        )
+        second = ScheduledNotificationFactory(
+            recipient=second_recipient,
+            section=self.section,
+            kind=NotificationKind.MEMBER_WAITING_DIGEST,
+            send_after=timezone.now() - timedelta(minutes=5),
+        )
+
+        def _side_effect(*, recipient_email, **kwargs):
+            if recipient_email == first_recipient.email:
+                raise Exception("SMTP exploded")
+            return
+
+        mock_send.side_effect = _side_effect
+
+        with self.assertLogs(
+            "apps.notifications.management.commands.send_scheduled_notifications",
+            level="ERROR",
+        ) as captured_logs:
+            call_command(COMMAND_NAME)
+
+        first.refresh_from_db()
+        second.refresh_from_db()
+
+        self.assertIsNone(first.sent_at)
+        self.assertIsNotNone(second.sent_at)
+        self.assertEqual(mock_send.call_count, 2)
+        self.assertIn("Failed to send scheduled notification pk=", "\n".join(captured_logs.output))
+
+    @patch("apps.notifications.management.commands.send_scheduled_notifications.send_notification_email")
+    def test_batch_size_limits_processed_notifications(self, mock_send):
+        """Batch size limits how many eligible notifications are sent in one run."""
+        first = self._make_pending()
+        second = self._make_pending()
+        third = self._make_pending()
+
+        call_command(COMMAND_NAME, batch_size=2)
+
+        first.refresh_from_db()
+        second.refresh_from_db()
+        third.refresh_from_db()
+
+        sent_count = ScheduledNotification.objects.filter(
+            pk__in=[first.pk, second.pk, third.pk],
+            sent_at__isnull=False,
+        ).count()
+        self.assertEqual(sent_count, 2)
+        self.assertEqual(mock_send.call_count, 2)
diff --git a/fiesta/apps/notifications/tests/test_scheduler.py b/fiesta/apps/notifications/tests/test_scheduler.py
index 1266dc86..c04e7935 100644
--- a/fiesta/apps/notifications/tests/test_scheduler.py
+++ b/fiesta/apps/notifications/tests/test_scheduler.py
@@ -5,7 +5,8 @@
 from django.test import TestCase
 from django.utils import timezone
 
-from apps.notifications.models import ScheduledNotification
+from apps.accounts.models import UserProfile
+from apps.notifications.models import NotificationKind, ScheduledNotification
 from apps.notifications.services.scheduler import cancel_scheduled_notifications_for, enqueue_delayed_notification
 from apps.notifications.tests.factories import ScheduledNotificationFactory
 from apps.sections.models import Section
@@ -13,16 +14,23 @@
 from apps.utils.factories.sections import KnownSectionFactory
 
 
+def _make_user_with_profile():
+    """Create a User with an attached UserProfile, avoiding the broken UserProfileFactory."""
+    user = UserFactory(profile=None)
+    UserProfile.objects.create(user=user)
+    return user
+
+
 class EnqueueDelayedNotificationTestCase(TestCase):
     def setUp(self):
-        self.user = UserFactory(profile=None)
+        self.user = _make_user_with_profile()
         self.section: Section = KnownSectionFactory()
         self.send_after = timezone.now() + timedelta(hours=1)
 
     def test_enqueue_creates_new_notification(self):
         """Calling enqueue creates a ScheduledNotification with the correct fields."""
         notification = enqueue_delayed_notification(
-            kind=ScheduledNotification.Kind.BUDDY_MATCHED_ISSUER,
+            kind=NotificationKind.BUDDY_MATCHED_ISSUER,
             recipient=self.user,
             section=self.section,
             related_object=self.section,  # use Section as a simple related_object
@@ -30,7 +38,7 @@ def test_enqueue_creates_new_notification(self):
         )
 
         self.assertIsNotNone(notification.pk)
-        self.assertEqual(notification.kind, ScheduledNotification.Kind.BUDDY_MATCHED_ISSUER)
+        self.assertEqual(notification.kind, NotificationKind.BUDDY_MATCHED_ISSUER)
         self.assertEqual(notification.recipient, self.user)
         self.assertEqual(notification.section, self.section)
         self.assertEqual(notification.send_after, self.send_after)
@@ -40,7 +48,7 @@ def test_enqueue_creates_new_notification(self):
     def test_enqueue_upserts_existing_pending(self):
         """Calling enqueue again for the same recipient+object updates send_after instead of creating a duplicate."""
         first = enqueue_delayed_notification(
-            kind=ScheduledNotification.Kind.BUDDY_MATCHED_ISSUER,
+            kind=NotificationKind.BUDDY_MATCHED_ISSUER,
             recipient=self.user,
             section=self.section,
             related_object=self.section,
@@ -49,7 +57,7 @@ def test_enqueue_upserts_existing_pending(self):
 
         later_time = self.send_after + timedelta(hours=2)
         second = enqueue_delayed_notification(
-            kind=ScheduledNotification.Kind.BUDDY_MATCHED_ISSUER,
+            kind=NotificationKind.BUDDY_MATCHED_ISSUER,
             recipient=self.user,
             section=self.section,
             related_object=self.section,
@@ -66,7 +74,7 @@ def test_enqueue_does_not_upsert_sent(self):
         """If an existing notification is already sent, a new one is created instead of updating."""
         # Create an already-sent notification
         sent_notification = ScheduledNotificationFactory(
-            kind=ScheduledNotification.Kind.BUDDY_MATCHED_ISSUER,
+            kind=NotificationKind.BUDDY_MATCHED_ISSUER,
             recipient=self.user,
             section=self.section,
             object_id=self.section.pk,
@@ -79,7 +87,7 @@ def test_enqueue_does_not_upsert_sent(self):
         sent_notification.save()
 
         new_notification = enqueue_delayed_notification(
-            kind=ScheduledNotification.Kind.BUDDY_MATCHED_ISSUER,
+            kind=NotificationKind.BUDDY_MATCHED_ISSUER,
             recipient=self.user,
             section=self.section,
             related_object=self.section,
@@ -97,7 +105,7 @@ def setUp(self):
 
     def test_cancel_cancels_pending(self):
         """cancel_scheduled_notifications_for marks pending notifications as cancelled."""
-        user = UserFactory(profile=None)
+        user = _make_user_with_profile()
         notification = ScheduledNotificationFactory(
             recipient=user,
             section=self.section,
@@ -116,7 +124,7 @@ def test_cancel_cancels_pending(self):
 
     def test_cancel_skips_already_sent(self):
         """cancel_scheduled_notifications_for does not touch notifications that are already sent."""
-        user = UserFactory(profile=None)
+        user = _make_user_with_profile()
         from django.contrib.contenttypes.models import ContentType
 
         notification = ScheduledNotificationFactory(
@@ -139,7 +147,7 @@ def test_cancel_returns_count(self):
         from django.contrib.contenttypes.models import ContentType
 
         ct = ContentType.objects.get_for_model(self.section)
-        users = [UserFactory(profile=None) for _ in range(3)]
+        users = [_make_user_with_profile() for _ in range(3)]
 
         for user in users:
             n = ScheduledNotificationFactory(
diff --git a/fiesta/apps/notifications/tests/test_unsubscribe.py b/fiesta/apps/notifications/tests/test_unsubscribe.py
new file mode 100644
index 00000000..b0fa3997
--- /dev/null
+++ b/fiesta/apps/notifications/tests/test_unsubscribe.py
@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+from django.conf import settings
+from django.core import mail
+from django.core.signing import BadSignature, SignatureExpired
+from django.test import TestCase
+
+from apps.accounts.models import UserProfile
+from apps.notifications.services.mailer import send_notification_email
+from apps.notifications.services.unsubscribe import generate_unsubscribe_token, verify_unsubscribe_token
+from apps.utils.factories.accounts import UserFactory
+
+
+class UnsubscribeTokenServiceTestCase(TestCase):
+    def test_generate_unsubscribe_token_returns_string(self):
+        token = generate_unsubscribe_token(user_id=123)
+        self.assertIsInstance(token, str)
+
+    def test_verify_unsubscribe_token_decodes_user_id_and_action(self):
+        token = generate_unsubscribe_token(user_id=321, action="global")
+
+        user_id, action = verify_unsubscribe_token(token)
+
+        self.assertEqual(user_id, "321")
+        self.assertEqual(action, "global")
+
+    def test_verify_unsubscribe_token_raises_signature_expired(self):
+        token = generate_unsubscribe_token(user_id=123)
+
+        with self.assertRaises(SignatureExpired):
+            verify_unsubscribe_token(token, max_age=-1)
+
+    def test_verify_unsubscribe_token_raises_bad_signature_for_tampered_token(self):
+        token = generate_unsubscribe_token(user_id=123)
+        tampered_token = f"{token}tampered"
+
+        with self.assertRaises(BadSignature):
+            verify_unsubscribe_token(tampered_token)
+
+
+class UnsubscribeViewTestCase(TestCase):
+    def setUp(self):
+        self.user = UserFactory(profile=None)
+        self.profile = UserProfile.objects.create(user=self.user)
+
+    def test_unsubscribe_post_sets_global_opt_out(self):
+        token = generate_unsubscribe_token(user_id=self.user.pk, action="global")
+
+        response = self.client.post(
+            f"/notifications/unsubscribe/{token}/",
+            SERVER_NAME=settings.ROOT_DOMAIN,
+        )
+
+        self.profile.refresh_from_db()
+        self.assertEqual(response.status_code, 200)
+        self.assertFalse(self.profile.email_notifications_enabled)
+
+    def test_unsubscribe_get_shows_confirmation_page(self):
+        token = generate_unsubscribe_token(user_id=self.user.pk, action="global")
+
+        response = self.client.get(
+            f"/notifications/unsubscribe/{token}/",
+            SERVER_NAME=settings.ROOT_DOMAIN,
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Unsubscribe from Notifications")
+
+    def test_unsubscribe_post_is_idempotent_for_already_opted_out_user(self):
+        self.profile.email_notifications_enabled = False
+        self.profile.save(update_fields=["email_notifications_enabled", "modified"])
+        token = generate_unsubscribe_token(user_id=self.user.pk, action="global")
+
+        response = self.client.post(
+            f"/notifications/unsubscribe/{token}/",
+            SERVER_NAME=settings.ROOT_DOMAIN,
+        )
+
+        self.profile.refresh_from_db()
+        self.assertEqual(response.status_code, 200)
+        self.assertFalse(self.profile.email_notifications_enabled)
+
+
+class NotificationMailerUnsubscribeHeadersTestCase(TestCase):
+    def test_sent_email_contains_list_unsubscribe_headers(self):
+        user = UserFactory(profile=None)
+        UserProfile.objects.create(user=user)
+
+        send_notification_email(
+            subject="Subject",
+            recipient_email=user.email,
+            template_prefix="notifications/base_email",
+            context={
+                "section": "ESN Test",
+                "preferences_url": "https://example.com/preferences",
+            },
+            recipient_user=user,
+        )
+
+        self.assertEqual(len(mail.outbox), 1)
+        sent_email = mail.outbox[0]
+
+        self.assertIn("List-Unsubscribe", sent_email.extra_headers)
+        self.assertIn("List-Unsubscribe-Post", sent_email.extra_headers)
+        self.assertEqual(sent_email.extra_headers["List-Unsubscribe-Post"], "List-Unsubscribe=One-Click")
+        self.assertIn("/notifications/unsubscribe/", sent_email.extra_headers["List-Unsubscribe"])
diff --git a/fiesta/apps/notifications/urls.py b/fiesta/apps/notifications/urls.py
index d93ea82c..c47c6738 100644
--- a/fiesta/apps/notifications/urls.py
+++ b/fiesta/apps/notifications/urls.py
@@ -2,10 +2,11 @@
 
 from django.urls import path
 
-from apps.notifications.views import NotificationPreferencesView
+from apps.notifications.views import NotificationPreferencesView, UnsubscribeView
 
 app_name = "notifications"
 
 urlpatterns = [
     path("preferences/", NotificationPreferencesView.as_view(), name="preferences"),
+    path("unsubscribe/<path:token>/", UnsubscribeView.as_view(), name="unsubscribe"),
 ]
diff --git a/fiesta/apps/notifications/views.py b/fiesta/apps/notifications/views.py
index 5c6bb6d2..a1ce4d21 100644
--- a/fiesta/apps/notifications/views.py
+++ b/fiesta/apps/notifications/views.py
@@ -1,15 +1,24 @@
 from __future__ import annotations
 
+import logging
 from typing import TYPE_CHECKING, Any
 
 from django.contrib.auth.mixins import LoginRequiredMixin
 from django.contrib.messages.views import SuccessMessageMixin
+from django.core.signing import BadSignature, SignatureExpired
+from django.http import HttpRequest, HttpResponse
+from django.shortcuts import render
+from django.utils.decorators import method_decorator
 from django.utils.translation import gettext_lazy as _
+from django.views import View
+from django.views.decorators.csrf import csrf_exempt
 from django.views.generic.edit import UpdateView
 
+from apps.accounts.models import UserProfile
 from apps.buddy_system.apps import BuddySystemConfig
 from apps.notifications.forms import NotificationPreferencesForm
-from apps.notifications.models import SectionNotificationPreferences
+from apps.notifications.models import NotificationKind, SectionNotificationPreferences
+from apps.notifications.services.unsubscribe import verify_unsubscribe_token
 from apps.pickup_system.apps import PickupSystemConfig
 from apps.plugins.views.mixins import CheckEnabledPluginsViewMixin
 from apps.sections.views.mixins.section_space import EnsureInSectionSpaceViewMixin
@@ -17,6 +26,8 @@
 if TYPE_CHECKING:
     from django.db.models import QuerySet
 
+logger = logging.getLogger(__name__)
+
 
 class NotificationPreferencesView(
     LoginRequiredMixin,
@@ -59,3 +70,114 @@ def get_form(self, form_class: type[NotificationPreferencesForm] | None = None)
         ):
             form.fields.pop("notify_on_match", None)
         return form
+
+
+@method_decorator(csrf_exempt, name="dispatch")
+class UnsubscribeView(View):
+    template_name = "notifications/unsubscribe.html"
+
+    def get(self, request: HttpRequest, token: str) -> HttpResponse:
+        try:
+            user_profile, action = self._resolve_token(token)
+        except UserProfile.DoesNotExist:
+            return render(
+                request,
+                self.template_name,
+                {"error": _("This unsubscribe link is invalid.")},
+                status=400,
+            )
+        except SignatureExpired:
+            return render(
+                request,
+                self.template_name,
+                {"error": _("This unsubscribe link has expired.")},
+                status=400,
+            )
+        except BadSignature:
+            return render(
+                request,
+                self.template_name,
+                {"error": _("This unsubscribe link is invalid.")},
+                status=400,
+            )
+
+        return render(
+            request,
+            self.template_name,
+            {
+                "action": action,
+                "action_description": self._action_description(action),
+                "email": user_profile.user.email,
+            },
+        )
+
+    def post(self, request: HttpRequest, token: str) -> HttpResponse:
+        try:
+            user_profile, action = self._resolve_token(token)
+            self._unsubscribe(user_profile=user_profile, action=action)
+        except UserProfile.DoesNotExist:
+            return render(
+                request,
+                self.template_name,
+                {"error": _("This unsubscribe link is invalid.")},
+                status=400,
+            )
+        except SignatureExpired:
+            return render(
+                request,
+                self.template_name,
+                {"error": _("This unsubscribe link has expired.")},
+                status=400,
+            )
+        except BadSignature:
+            return render(
+                request,
+                self.template_name,
+                {"error": _("This unsubscribe link is invalid.")},
+                status=400,
+            )
+
+        return render(
+            request,
+            self.template_name,
+            {
+                "success": True,
+                "action": action,
+                "action_description": self._action_description(action),
+            },
+        )
+
+    def _resolve_token(self, token: str) -> tuple[UserProfile, str]:
+        user_id, action = verify_unsubscribe_token(token)
+        return UserProfile.objects.select_related("user").get(user_id=user_id), action
+
+    def _unsubscribe(self, *, user_profile: UserProfile, action: str) -> None:
+        if action == "global":
+            if user_profile.email_notifications_enabled:
+                user_profile.email_notifications_enabled = False
+                user_profile.save(update_fields=["email_notifications_enabled", "modified"])
+            return
+
+        if action in {NotificationKind.BUDDY_MATCHED_ISSUER, NotificationKind.PICKUP_MATCHED_ISSUER}:
+            SectionNotificationPreferences.objects.filter(user=user_profile.user).update(notify_on_match=False)
+            return
+
+        if action == NotificationKind.MEMBER_WAITING_DIGEST:
+            SectionNotificationPreferences.objects.filter(user=user_profile.user).update(
+                notify_on_new_member_waiting=False
+            )
+            return
+
+        logger.warning("Unsupported unsubscribe action %r for user_profile=%s", action, user_profile.pk)
+
+    def _action_description(self, action: str) -> str:
+        if action == "global":
+            return _("all email notifications")
+
+        if action in {NotificationKind.BUDDY_MATCHED_ISSUER, NotificationKind.PICKUP_MATCHED_ISSUER}:
+            return _("match notification emails")
+
+        if action == NotificationKind.MEMBER_WAITING_DIGEST:
+            return _("new member waiting digest emails")
+
+        return _("selected email notifications")