diff --git a/app/job/rest.py b/app/job/rest.py
index 28bfeafd33..eb0c41dbe8 100644
--- a/app/job/rest.py
+++ b/app/job/rest.py
@@ -33,7 +33,9 @@
 )
 from app.notifications.process_notifications import simulated_recipient
 from app.notifications.validators import (
+    check_email_annual_limit,
     check_email_daily_limit,
+    check_sms_annual_limit,
     check_sms_daily_limit,
     increment_email_daily_count_send_warnings_if_needed,
     increment_sms_daily_count_send_warnings_if_needed,
@@ -183,6 +185,7 @@ def create_job(service_id):
         is_test_notification = len(recipient_csv) == numberOfSimulated
 
         if not is_test_notification:
+            check_sms_annual_limit(service, len(recipient_csv))
             check_sms_daily_limit(service, len(recipient_csv))
             increment_sms_daily_count_send_warnings_if_needed(service, len(recipient_csv))
 
@@ -195,6 +198,7 @@ def create_job(service_id):
             )
             notification_count = len(recipient_csv)
 
+        check_email_annual_limit(service, notification_count)
         check_email_daily_limit(service, notification_count)
 
         scheduled_for = datetime.fromisoformat(data.get("scheduled_for")) if data.get("scheduled_for") else None
diff --git a/app/notifications/rest.py b/app/notifications/rest.py
index e3ec327df6..053014ec81 100644
--- a/app/notifications/rest.py
+++ b/app/notifications/rest.py
@@ -22,8 +22,10 @@
     simulated_recipient,
 )
 from app.notifications.validators import (
+    check_email_annual_limit,
     check_email_daily_limit,
     check_rate_limiting,
+    check_sms_annual_limit,
     check_template_is_active,
     check_template_is_for_notification_type,
     service_has_permission,
@@ -113,6 +115,7 @@ def send_notification(notification_type: NotificationType):
 
     simulated = simulated_recipient(notification_form["to"], notification_type)
     if not simulated != api_user.key_type == KEY_TYPE_TEST and notification_type == EMAIL_TYPE:
+        check_email_annual_limit(authenticated_service, 1)
         check_email_daily_limit(authenticated_service, 1)
 
     check_template_is_for_notification_type(notification_type, template.template_type)
@@ -129,6 +132,8 @@ def send_notification(notification_type: NotificationType):
 
     if notification_type == SMS_TYPE:
         _service_can_send_internationally(authenticated_service, notification_form["to"])
+        if not simulated and api_user.key_type != KEY_TYPE_TEST:
+            check_sms_annual_limit(authenticated_service, 1)
     # Do not persist or send notification to the queue if it is a simulated recipient
 
     notification_model = persist_notification(
diff --git a/app/notifications/validators.py b/app/notifications/validators.py
index 0ff52eb924..229552ee8c 100644
--- a/app/notifications/validators.py
+++ b/app/notifications/validators.py
@@ -218,7 +218,7 @@ def check_email_annual_limit(service: Service, requested_emails=0):
         return
 
     current_app.logger.info(
-        f"Service {service.id} is exceeding their annual email limit [total sent this fiscal: {int(emails_sent_today + emails_sent_this_fiscal)} limit: {service.email_annual_limit}, attempted send: {requested_emails}"
+        f"{'Trial service' if service.restricted else 'Service'} {service.id} is exceeding their annual email limit [total sent this fiscal: {int(emails_sent_today + emails_sent_this_fiscal)} limit: {service.email_annual_limit}, attempted send: {requested_emails}"
     )
     if service.restricted:
         raise TrialServiceRequestExceedsEmailAnnualLimitError(service.email_annual_limit)
@@ -549,6 +549,7 @@ def send_annual_limit_updated_email(service: Service, notification_type: Notific
         service_id=service.id,
         template_id=current_app.config["ANNUAL_LIMIT_UPDATED_TEMPLATE_ID"],
         personalisation={
+            "service_name": service.name,
             "message_type_en": notification_type,
             "message_type_fr": "Courriel" if notification_type == EMAIL_TYPE else "SMS",
             "message_limit_en": service.email_annual_limit if notification_type == EMAIL_TYPE else service.sms_annual_limit,
diff --git a/app/service/send_notification.py b/app/service/send_notification.py
index 77cd1313e3..40d260a1e6 100644
--- a/app/service/send_notification.py
+++ b/app/service/send_notification.py
@@ -33,9 +33,11 @@
     simulated_recipient,
 )
 from app.notifications.validators import (
+    check_email_annual_limit,
     check_email_daily_limit,
     check_service_has_permission,
     check_service_over_daily_message_limit,
+    check_sms_annual_limit,
     check_sms_daily_limit,
     increment_email_daily_count_send_warnings_if_needed,
     increment_sms_daily_count_send_warnings_if_needed,
@@ -70,8 +72,10 @@ def send_one_off_notification(service_id, post_data):
     if template.template_type == SMS_TYPE:
         is_test_notification = simulated_recipient(post_data["to"], template.template_type)
         if not is_test_notification:
+            check_sms_annual_limit(service, 1)
             check_sms_daily_limit(service, 1)
     elif template.template_type == EMAIL_TYPE:
+        check_email_annual_limit(service, 1)
         check_email_daily_limit(service, 1)  # 1 email
 
     validate_and_format_recipient(
diff --git a/app/v2/notifications/post_notifications.py b/app/v2/notifications/post_notifications.py
index d8edf8627b..537bc2849a 100644
--- a/app/v2/notifications/post_notifications.py
+++ b/app/v2/notifications/post_notifications.py
@@ -62,7 +62,6 @@
     Notification,
     NotificationType,
     Service,
-    TemplateType,
 )
 from app.notifications.process_letter_notifications import create_letter_notification
 from app.notifications.process_notifications import (
@@ -74,12 +73,14 @@
     transform_notification,
 )
 from app.notifications.validators import (
+    check_email_annual_limit,
     check_email_daily_limit,
     check_rate_limiting,
     check_service_can_schedule_notification,
     check_service_email_reply_to_id,
     check_service_has_permission,
     check_service_sms_sender_id,
+    check_sms_annual_limit,
     check_sms_daily_limit,
     increment_email_daily_count_send_warnings_if_needed,
     increment_sms_daily_count_send_warnings_if_needed,
@@ -183,10 +184,14 @@ def post_bulk():
 
     if template.template_type == SMS_TYPE:
         fragments_sent = fetch_todays_requested_sms_count(authenticated_service.id)
-        remaining_messages = authenticated_service.sms_daily_limit - fragments_sent
+        remaining_daily_messages = authenticated_service.sms_daily_limit - fragments_sent
+        remaining_annual_messages = authenticated_service.sms_annual_limit - fragments_sent
+
     else:
         current_app.logger.info(f"[post_notifications.post_bulk()] Checking bounce rate for service: {authenticated_service.id}")
-        remaining_messages = authenticated_service.message_limit - fetch_todays_total_message_count(authenticated_service.id)
+        emails_sent = fetch_todays_total_message_count(authenticated_service.id)
+        remaining_daily_messages = authenticated_service.message_limit - emails_sent
+        remaining_annual_messages = authenticated_service.email_annual_limit - emails_sent
 
     form["validated_sender_id"] = validate_sender_id(template, form.get("reply_to_id"))
 
@@ -199,19 +204,33 @@ def post_bulk():
         else:
             file_data = form["csv"]
 
-        recipient_csv = RecipientCSV(
-            file_data,
-            template_type=template.template_type,
-            placeholders=template._as_utils_template().placeholders,
-            max_rows=max_rows,
-            safelist=safelisted_members(authenticated_service, api_user.key_type),
-            remaining_messages=remaining_messages,
-            template=Template(template.__dict__),
-        )
+        if current_app.config["FF_ANNUAL_LIMIT"]:
+            recipient_csv = RecipientCSV(
+                file_data,
+                template_type=template.template_type,
+                placeholders=template._as_utils_template().placeholders,
+                max_rows=max_rows,
+                safelist=safelisted_members(authenticated_service, api_user.key_type),
+                remaining_messages=remaining_daily_messages,
+                remaining_daily_messages=remaining_daily_messages,
+                remaining_annual_messages=remaining_annual_messages,
+                template=Template(template.__dict__),
+            )
+        else:  # TODO FF_ANNUAL_LIMIT REMOVAL - Remove this block
+            recipient_csv = RecipientCSV(
+                file_data,
+                template_type=template.template_type,
+                placeholders=template._as_utils_template().placeholders,
+                max_rows=max_rows,
+                safelist=safelisted_members(authenticated_service, api_user.key_type),
+                remaining_messages=remaining_daily_messages,
+                template=Template(template.__dict__),
+            )
     except csv.Error as e:
         raise BadRequestError(message=f"Error converting to CSV: {str(e)}", status_code=400)
 
-    check_for_csv_errors(recipient_csv, max_rows, remaining_messages)
+    check_for_csv_errors(recipient_csv, max_rows, remaining_daily_messages, remaining_annual_messages)
+    notification_count_requested = len(list(recipient_csv.get_rows()))
 
     for row in recipient_csv.get_rows():
         try:
@@ -223,11 +242,12 @@ def post_bulk():
             raise BadRequestError(message=message)
 
     if template.template_type == EMAIL_TYPE and api_user.key_type != KEY_TYPE_TEST:
-        check_email_daily_limit(authenticated_service, len(list(recipient_csv.get_rows())))
+        check_email_annual_limit(authenticated_service, notification_count_requested)
+        check_email_daily_limit(authenticated_service, notification_count_requested)
         scheduled_for = datetime.fromisoformat(form.get("scheduled_for")) if form.get("scheduled_for") else None
 
         if scheduled_for is None or not scheduled_for.date() > datetime.today().date():
-            increment_email_daily_count_send_warnings_if_needed(authenticated_service, len(list(recipient_csv.get_rows())))
+            increment_email_daily_count_send_warnings_if_needed(authenticated_service, notification_count_requested)
 
     if template.template_type == SMS_TYPE:
         # set sender_id if missing
@@ -240,15 +260,16 @@ def post_bulk():
         numberOfSimulated = sum(
             simulated_recipient(i["phone_number"].data, template.template_type) for i in list(recipient_csv.get_rows())
         )
-        mixedRecipients = numberOfSimulated > 0 and numberOfSimulated != len(list(recipient_csv.get_rows()))
+        mixedRecipients = numberOfSimulated > 0 and numberOfSimulated != notification_count_requested
 
         # if its a live or a team key, and they have specified testing and NON-testing recipients, raise an error
         if api_user.key_type != KEY_TYPE_TEST and mixedRecipients:
             raise BadRequestError(message="Bulk sending to testing and non-testing numbers is not supported", status_code=400)
 
-        is_test_notification = api_user.key_type == KEY_TYPE_TEST or len(list(recipient_csv.get_rows())) == numberOfSimulated
+        is_test_notification = api_user.key_type == KEY_TYPE_TEST or notification_count_requested == numberOfSimulated
 
         if not is_test_notification:
+            check_sms_annual_limit(authenticated_service, len(recipient_csv))
             check_sms_daily_limit(authenticated_service, len(recipient_csv))
             increment_sms_daily_count_send_warnings_if_needed(authenticated_service, len(recipient_csv))
 
@@ -302,11 +323,13 @@ def post_notification(notification_type: NotificationType):
     )
 
     if template.template_type == EMAIL_TYPE and api_user.key_type != KEY_TYPE_TEST:
+        check_email_annual_limit(authenticated_service, 1)
         check_email_daily_limit(authenticated_service, 1)  # 1 email
 
     if template.template_type == SMS_TYPE:
         is_test_notification = api_user.key_type == KEY_TYPE_TEST or simulated_recipient(form["phone_number"], notification_type)
         if not is_test_notification:
+            check_sms_annual_limit(authenticated_service, 1)
             check_sms_daily_limit(authenticated_service, 1)
 
     current_app.logger.info(f"Trying to send notification for Template ID: {template.id}")
@@ -664,7 +687,7 @@ def strip_keys_from_personalisation_if_send_attach(personalisation):
     return {k: v for (k, v) in personalisation.items() if not (type(v) is dict and v.get("sending_method") == "attach")}
 
 
-def check_for_csv_errors(recipient_csv, max_rows, remaining_messages):
+def check_for_csv_errors(recipient_csv, max_rows, remaining_daily_messages, remaining_annual_messages):
     nb_rows = len(recipient_csv)
 
     if recipient_csv.has_errors:
@@ -678,15 +701,38 @@ def check_for_csv_errors(recipient_csv, max_rows, remaining_messages):
                 message=f"Duplicate column headers: {', '.join(sorted(recipient_csv.duplicate_recipient_column_headers))}",
                 status_code=400,
             )
+        ## TODO: FF_ANNUAL_LIMIT - remove this if block in favour of more_rows_than_can_send_today found below
         if recipient_csv.more_rows_than_can_send:
             if recipient_csv.template_type == SMS_TYPE:
                 raise BadRequestError(
-                    message=f"You only have {remaining_messages} remaining sms messages before you reach your daily limit. You've tried to send {len(recipient_csv)} sms messages.",
+                    message=f"You only have {remaining_daily_messages} remaining sms messages before you reach your daily limit. You've tried to send {len(recipient_csv)} sms messages.",
+                    status_code=400,
+                )
+            else:
+                raise BadRequestError(
+                    message=f"You only have {remaining_daily_messages} remaining messages before you reach your daily limit. You've tried to send {nb_rows} messages.",
+                    status_code=400,
+                )
+        if recipient_csv.more_rows_than_can_send_this_year:
+            if recipient_csv.template_type == SMS_TYPE:
+                raise BadRequestError(
+                    message=f"You only have {remaining_annual_messages} remaining sms messages before you reach your annual limit. You've tried to send {len(recipient_csv)} sms messages.",
+                    status_code=400,
+                )
+            else:
+                raise BadRequestError(
+                    message=f"You only have {remaining_annual_messages} remaining messages before you reach your annual limit. You've tried to send {nb_rows} messages.",
+                    status_code=400,
+                )
+        if recipient_csv.more_rows_than_can_send_today:
+            if recipient_csv.template_type == SMS_TYPE:
+                raise BadRequestError(
+                    message=f"You only have {remaining_daily_messages} remaining sms messages before you reach your daily limit. You've tried to send {len(recipient_csv)} sms messages.",
                     status_code=400,
                 )
             else:
                 raise BadRequestError(
-                    message=f"You only have {remaining_messages} remaining messages before you reach your daily limit. You've tried to send {nb_rows} messages.",
+                    message=f"You only have {remaining_daily_messages} remaining messages before you reach your daily limit. You've tried to send {nb_rows} messages.",
                     status_code=400,
                 )
 
diff --git a/poetry.lock b/poetry.lock
index f2936fae5a..1306f3961a 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -2682,7 +2682,7 @@ requests = ">=2.0.0"
 
 [[package]]
 name = "notifications-utils"
-version = "52.4.0"
+version = "52.4.1"
 description = "Shared python code for Notification - Provides logging utils etc."
 optional = false
 python-versions = "~3.10.9"
@@ -2718,8 +2718,8 @@ werkzeug = "3.0.4"
 [package.source]
 type = "git"
 url = "https://github.com/cds-snc/notifier-utils.git"
-reference = "52.4.0"
-resolved_reference = "25ef1ae7703ec68b622bf24c4ea08f20ded0bab3"
+reference = "task/pre-3.12-recipient-csv-annual-limit-validation"
+resolved_reference = "f76351fbb28fd832c87016c7cc7ce69276ee88d5"
 
 [[package]]
 name = "ordered-set"
@@ -4682,4 +4682,4 @@ testing = ["coverage (>=5.0.3)", "zope.event", "zope.testing"]
 [metadata]
 lock-version = "2.0"
 python-versions = "~3.10.9"
-content-hash = "4b6cd477a14779de5ae5399a00bb936b487d7e288564ab969456029ad74d059f"
+content-hash = "0c8c24f0b4bdb8b565d8c711fd214bb7dbbe78b49de39e9cd5dcbd006512545c"
diff --git a/pyproject.toml b/pyproject.toml
index f9c0d35e7e..a60a709e8d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -65,7 +65,7 @@ Werkzeug = "3.0.4"
 MarkupSafe = "2.1.5"
 # REVIEW: v2 is using sha512 instead of sha1 by default (in v1)
 itsdangerous = "2.2.0"
-notifications-utils = { git = "https://github.com/cds-snc/notifier-utils.git", tag = "52.4.0" }
+notifications-utils = { git = "https://github.com/cds-snc/notifier-utils.git", branch = "task/pre-3.12-recipient-csv-annual-limit-validation"}
 
 # rsa = "4.9  # awscli 1.22.38 depends on rsa<4.8
 typing-extensions = "4.12.2"
diff --git a/tests/app/db.py b/tests/app/db.py
index 664445fbcc..1d2c2c56d3 100644
--- a/tests/app/db.py
+++ b/tests/app/db.py
@@ -112,8 +112,8 @@ def create_service(
     prefix_sms=True,
     message_limit=1000,
     sms_daily_limit=1000,
-    annual_email_limit=10000000,
-    annual_sms_limit=25000,
+    email_annual_limit=10000000,
+    sms_annual_limit=25000,
     organisation_type="central",
     check_if_service_exists=False,
     go_live_user=None,
@@ -136,6 +136,8 @@ def create_service(
             organisation_type=organisation_type,
             go_live_user=go_live_user,
             go_live_at=go_live_at,
+            email_annual_limit=email_annual_limit,
+            sms_annual_limit=sms_annual_limit,
             crown=crown,
             sensitive_service=sensitive_service,
         )
diff --git a/tests/app/job/test_rest.py b/tests/app/job/test_rest.py
index 6a49c8ce63..1574303bfe 100644
--- a/tests/app/job/test_rest.py
+++ b/tests/app/job/test_rest.py
@@ -9,6 +9,12 @@
 import app.celery.tasks
 from app.dao.templates_dao import dao_update_template
 from app.models import JOB_STATUS_PENDING, JOB_STATUS_TYPES, ServiceSmsSender
+from app.notifications.validators import (
+    LiveServiceRequestExceedsEmailAnnualLimitError,
+    LiveServiceRequestExceedsSMSAnnualLimitError,
+    TrialServiceRequestExceedsEmailAnnualLimitError,
+    TrialServiceRequestExceedsSMSAnnualLimitError,
+)
 from tests import create_authorization_header
 from tests.app.db import (
     create_ft_notification_status,
@@ -624,6 +630,61 @@ def test_create_job_returns_400_if_archived_template(client, sample_template, mo
     assert "Template has been deleted" in resp_json["message"]["template"]
 
 
+@pytest.mark.parametrize(
+    "template_type, exception",
+    [
+        ("sms", LiveServiceRequestExceedsSMSAnnualLimitError),
+        ("sms", TrialServiceRequestExceedsSMSAnnualLimitError),
+        ("email", LiveServiceRequestExceedsEmailAnnualLimitError),
+        ("email", TrialServiceRequestExceedsEmailAnnualLimitError),
+    ],
+)
+def test_create_job_should_429_when_over_annual_limit(
+    client,
+    mocker,
+    sample_template,
+    sample_email_template,
+    fake_uuid,
+    template_type,
+    exception,
+):
+    template = sample_template if template_type == "sms" else sample_email_template
+    email_to = template.service.created_by.email_address if template_type == "email" else None
+    limit = template.service.sms_annual_limit if template_type == "sms" else template.service.email_annual_limit
+    path = "/service/{}/job".format(template.service.id)
+    mocker.patch(
+        "app.job.rest.get_job_metadata_from_s3",
+        return_value={
+            "template_id": str(template.id),
+            "original_file_name": "thisisatest.csv",
+            "notification_count": "2",
+            "valid": "True",
+        },
+    )
+    mocker.patch(
+        "app.job.rest.get_job_from_s3",
+        return_value="phone number\r\n6502532222\r\n6502532222"
+        if template_type == "sms"
+        else f"email address\r\n{email_to}\r\n{email_to}",
+    )
+
+    mocker.patch(f"app.job.rest.check_{template_type}_annual_limit", side_effect=exception(limit))
+    data = {
+        "id": fake_uuid,
+        "created_by": str(template.created_by.id),
+    }
+    auth_header = create_authorization_header()
+    headers = [("Content-Type", "application/json"), auth_header]
+    response = client.post(path, data=json.dumps(data), headers=headers)
+
+    resp_json = json.loads(response.get_data(as_text=True))
+    assert response.status_code == 429
+    assert (
+        resp_json["message"]
+        == f"Exceeded annual {template_type if template_type == 'email' else template_type.upper()} sending limit of {limit} messages"
+    )
+
+
 def _setup_jobs(template, number_of_jobs=5):
     for i in range(number_of_jobs):
         create_job(template=template)
diff --git a/tests/app/notifications/rest/test_send_notification.py b/tests/app/notifications/rest/test_send_notification.py
index a1cf1f0d52..59d239999a 100644
--- a/tests/app/notifications/rest/test_send_notification.py
+++ b/tests/app/notifications/rest/test_send_notification.py
@@ -45,6 +45,7 @@
     create_sample_template_without_sms_permission,
 )
 from tests.app.db import create_reply_to_email, create_service
+from tests.conftest import set_config
 
 
 @pytest.mark.parametrize("template_type", [SMS_TYPE, EMAIL_TYPE])
@@ -384,6 +385,75 @@ def test_should_allow_valid_email_notification(notify_api, sample_email_template
             assert response_data["template_version"] == sample_email_template.version
 
 
+@pytest.mark.parametrize(
+    "create_template_func, payload, expected_error_message, annual_limit",
+    [
+        (
+            create_sample_email_template,
+            {"to": "ok@ok.com", "template": "", "valid": "True"},
+            "Exceeded annual email sending limit of 1 messages",
+            1,
+        ),
+        (
+            create_sample_template,
+            {"to": "+16502532222", "template": "", "valid": "True"},
+            "Exceeded annual SMS sending limit of 1 messages",
+            1,
+        ),
+        (create_sample_email_template, {"to": "ok@ok.com", "template": "", "valid": "True"}, None, 2),
+        (create_sample_template, {"to": "+16502532222", "template": "", "valid": "True"}, None, 2),
+    ],
+)
+@pytest.mark.parametrize("is_trial", [True, False])
+def test_should_block_api_call_if_over_annual_limit_and_allow_if_under_limit(
+    notify_db,
+    notify_db_session,
+    notify_api,
+    mocker,
+    is_trial,
+    create_template_func,
+    payload,
+    expected_error_message,
+    annual_limit,
+):
+    with notify_api.test_request_context(), set_config(notify_api, "FF_ANNUAL_LIMIT", True):
+        with notify_api.test_client() as client:
+            service = create_sample_service(
+                notify_db, notify_db_session, sms_annual_limit=annual_limit, email_annual_limit=annual_limit, restricted=is_trial
+            )
+            template = create_template_func(notify_db, notify_db_session, service=service)
+            payload["template"] = str(template.id)
+            payload["to"] = (
+                template.service.created_by.email_address if is_trial and template.template_type == EMAIL_TYPE else payload["to"]
+            )
+
+            mocker.patch("app.notifications.validators.check_service_over_api_rate_limit_and_update_rate")
+            mocker.patch(f"app.celery.provider_tasks.deliver_{template.template_type}.apply_async")
+            mocker.patch("app.notifications.validators.send_notification_to_service_users")
+
+            create_sample_notification(
+                notify_db,
+                notify_db_session,
+                template=template,
+                service=service,
+                created_at=datetime.utcnow(),
+            )
+
+            auth_header = create_authorization_header(service_id=service.id)
+            response = client.post(
+                path=f"/notifications/{template.template_type}",
+                data=json.dumps(payload),
+                headers=[("Content-Type", "application/json"), auth_header],
+            )
+            message = json.loads(response.get_data(as_text=True))
+
+            if expected_error_message:
+                assert response.status_code == 429
+                assert message["message"] == expected_error_message
+            else:
+                assert response.status_code == 201
+
+
 @freeze_time("2016-01-01 12:00:00.061258")
 def test_should_block_api_call_if_over_day_limit_for_live_service(notify_db, notify_db_session, notify_api, mocker):
     with notify_api.test_request_context():
diff --git a/tests/app/v2/notifications/test_post_notifications.py b/tests/app/v2/notifications/test_post_notifications.py
index cc33c4d527..bbd2b100c6 100644
--- a/tests/app/v2/notifications/test_post_notifications.py
+++ b/tests/app/v2/notifications/test_post_notifications.py
@@ -482,6 +482,58 @@ def test_returns_a_429_limit_exceeded_if_rate_limit_exceeded(
 
         assert not save_mock.called
 
+    @pytest.mark.parametrize(
+        "notification_type, key_send_to, send_to, expected_error_message",
+        [
+            ("sms", "phone_number", "6502532222", "Exceeded annual SMS sending limit of 1 messages"),
+            ("email", "email_address", "sample@email.com", "Exceeded annual email sending limit of 1 messages"),
+        ],
+    )
+    def test_post_notification_returns_429_when_annual_limit_exceeded(
+        self,
+        notify_db,
+        notify_db_session,
+        notify_api,
+        client,
+        sample_service,
+        mocker,
+        notification_type,
+        key_send_to,
+        send_to,
+        expected_error_message,
+    ):
+        sample_service.sms_annual_limit = 1
+        sample_service.email_annual_limit = 1
+        template = create_template(service=sample_service, template_type=notification_type)
+        save_mock = mocker.patch("app.v2.notifications.post_notifications.db_save_and_send_notification")
+        mocker.patch("app.v2.notifications.post_notifications.check_rate_limiting")
+
+        data = {key_send_to: send_to, "template_id": str(template.id)}
+
+        auth_header = create_authorization_header(service_id=template.service_id)
+
+        with set_config(notify_api, "FF_ANNUAL_LIMIT", True):
+            # Success
+            create_sample_notification(
+                notify_db,
+                notify_db_session,
+                template=template,
+                service=sample_service,
+                created_at=datetime.utcnow(),
+            )
+            with set_config(notify_api, "FF_ANNUAL_LIMIT", True):
+                response = client.post(
+                    path="/v2/notifications/{}".format(notification_type),
+                    data=json.dumps(data),
+                    headers=[("Content-Type", "application/json"), auth_header],
+                )
+            message = json.loads(response.data)["errors"][0]["message"]
+            status_code = json.loads(response.data)["status_code"]
+            assert status_code == 429
+            assert message == expected_error_message
+
+            assert not save_mock.called
+
     def test_post_sms_notification_returns_400_if_not_allowed_to_send_int_sms(
         self,
         client,
@@ -2304,7 +2356,7 @@ def test_post_bulk_flags_recipient_not_in_safelist_with_restricted_service(self,
             }
         ]
 
-    def test_post_bulk_flags_not_enough_remaining_messages(self, client, notify_db, notify_db_session, mocker):
+    def test_post_bulk_flags_not_enough_remaining_messages(self, client, notify_api, notify_db, notify_db_session, mocker):
         service = create_service(message_limit=10)
         template = create_sample_template(notify_db, notify_db_session, service=service, template_type="email")
         messages_count_mock = mocker.patch(
@@ -2316,21 +2368,23 @@ def test_post_bulk_flags_not_enough_remaining_messages(self, client, notify_db,
             "csv": rows_to_csv([["email address"], ["foo@example.com"], ["bar@example.com"]]),
         }
 
-        response = client.post(
-            "/v2/notifications/bulk",
-            data=json.dumps(data),
-            headers=[("Content-Type", "application/json"), create_authorization_header(service_id=template.service_id)],
-        )
+        # TODO: FF_ANNUAL_LIMIT removal
+        with set_config(notify_api, "FF_ANNUAL_LIMIT", True):
+            response = client.post(
+                "/v2/notifications/bulk",
+                data=json.dumps(data),
+                headers=[("Content-Type", "application/json"), create_authorization_header(service_id=template.service_id)],
+            )
 
-        assert response.status_code == 400
-        error_json = json.loads(response.get_data(as_text=True))
-        assert error_json["errors"] == [
-            {
-                "error": "BadRequestError",
-                "message": "You only have 1 remaining messages before you reach your daily limit. You've tried to send 2 messages.",
-            }
-        ]
-        messages_count_mock.assert_called_once()
+            assert response.status_code == 400
+            error_json = json.loads(response.get_data(as_text=True))
+            assert error_json["errors"] == [
+                {
+                    "error": "BadRequestError",
+                    "message": "You only have 1 remaining messages before you reach your daily limit. You've tried to send 2 messages.",
+                }
+            ]
+            messages_count_mock.assert_called_once()
 
     def test_post_bulk_flags_not_enough_remaining_sms_messages(self, notify_api, client, notify_db, notify_db_session, mocker):
         service = create_service(sms_daily_limit=10, message_limit=100)
@@ -2345,21 +2399,23 @@ def test_post_bulk_flags_not_enough_remaining_sms_messages(self, notify_api, cli
             "csv": rows_to_csv([["phone number"], ["6135551234"], ["6135551234"]]),
         }
 
-        response = client.post(
-            "/v2/notifications/bulk",
-            data=json.dumps(data),
-            headers=[("Content-Type", "application/json"), create_authorization_header(service_id=template.service_id)],
-        )
+        # TODO: FF_ANNUAL_LIMIT removal
+        with set_config(notify_api, "FF_ANNUAL_LIMIT", True):
+            response = client.post(
+                "/v2/notifications/bulk",
+                data=json.dumps(data),
+                headers=[("Content-Type", "application/json"), create_authorization_header(service_id=template.service_id)],
+            )
 
-        assert response.status_code == 400
-        error_json = json.loads(response.get_data(as_text=True))
-        assert error_json["errors"] == [
-            {
-                "error": "BadRequestError",
-                "message": "You only have 1 remaining sms messages before you reach your daily limit. You've tried to send 2 sms messages.",
-            }
-        ]
-        messages_count_mock.assert_called_once()
+            assert response.status_code == 400
+            error_json = json.loads(response.get_data(as_text=True))
+            assert error_json["errors"] == [
+                {
+                    "error": "BadRequestError",
+                    "message": "You only have 1 remaining sms messages before you reach your daily limit. You've tried to send 2 sms messages.",
+                }
+            ]
+            messages_count_mock.assert_called_once()
 
     @pytest.mark.parametrize("data_type", ["rows", "csv"])
     def test_post_bulk_flags_rows_with_errors(self, client, notify_db, notify_db_session, data_type):
@@ -2643,6 +2699,109 @@ def test_email_each_queue_is_used(self, notify_api, client, mocker, service_fact
             assert mock_redisQueue_EMAIL_PRIORITY.called
 
 
+@pytest.mark.parametrize(
+    "template_type, payload, expected_error_message, annual_limit",
+    [
+        (
+            "email",
+            {"to": "ok@ok.com", "valid": "True"},
+            "Exceeded annual email sending limit of 2 messages",
+            2,
+        ),
+        (
+            "sms",
+            {"to": "+16502532222", "valid": "True"},
+            "Exceeded annual SMS sending limit of 2 messages",
+            2,
+        ),
+    ],
+)
+@pytest.mark.parametrize("restricted", [True, False])
+def test_API_one_off_sends_blocks_sends_when_over_annual_limit_allows_if_under_limit(
+    notify_api,
+    client,
+    notify_db,
+    notify_db_session,
+    mocker,
+    template_type,
+    payload,
+    restricted,
+    expected_error_message,
+    annual_limit,
+):
+    # test setup
+    mocker.patch("app.sms_normal_publish.publish")
+    mocker.patch("app.service.send_notification.send_notification_to_queue")
+    mocker.patch("app.notifications.validators.service_can_send_to_recipient")
+    mocker.patch("app.notifications.validators.send_notification_to_service_users")
+
+    def __send_notification():
+        with set_config_values(notify_api, {"REDIS_ENABLED": True, "FF_ANNUAL_LIMIT": True}):
+            token = create_jwt_token(
+                current_app.config["ADMIN_CLIENT_SECRET"], client_id=current_app.config["ADMIN_CLIENT_USER_NAME"]
+            )
+            response = client.post(
+                f"/service/{template.service_id}/send-notification",
+                json=payload,
+                headers={"Authorization": f"Bearer {token}"},
+            )
+        return response
+
+    service = create_service(sms_annual_limit=annual_limit, restricted=restricted, email_annual_limit=annual_limit)
+    template = create_sample_template(notify_db, notify_db_session, content="Hello", service=service, template_type=template_type)
+    payload["template_id"] = template.id
+    payload["created_by"] = service.users[0].id
+
+    for x in range(annual_limit - 1):
+        create_sample_notification(notify_db, notify_db_session, to_field=payload["to"], template=template, service=service)
+
+    assert __send_notification().status_code == 201  # Ensure send is allowed while under the limit
+    response = __send_notification()  # Attempt to go over
+    assert response.status_code == 429  # Ensure send is blocked
+    assert json.loads(response.get_data(as_text=True))["message"] == expected_error_message
+
+
+@pytest.mark.parametrize(
+    "template_type, expected_msg",
+    [
+        ("email", "You only have 1 remaining messages before you reach your annual limit. You've tried to send 4 messages."),
+        (
+            "sms",
+            "You only have 1 remaining sms messages before you reach your annual limit. You've tried to send 4 sms messages.",
+        ),
+    ],
+)
+@pytest.mark.parametrize("data_type", ["rows", "csv"])
+def test_post_bulk_validates_annual_limit(
+    notify_api, notify_db, notify_db_session, client, template_type, data_type, expected_msg, mocker
+):
+    service = create_service(email_annual_limit=1, sms_annual_limit=1)
+    template = create_template(service=service, template_type=template_type)
+    data = {
+        "name": "job_name",
+        "template_id": template.id,
+    }
+    to_heading = "email address" if template_type == "email" else "phone number"
+    to = service.created_by.email_address if template_type == "email" else "6502532222"
+    rows = [[to_heading], *[to for _ in range(4)]]
+    if data_type == "csv":
+        data["csv"] = rows_to_csv(rows)
+    else:
+        data["rows"] = rows
+
+    auth_header = create_authorization_header(service_id=template.service.id)
+
+    with set_config(notify_api, "FF_ANNUAL_LIMIT", True):
+        response = client.post(
+            path="/v2/notifications/bulk",
+            data=json.dumps(data),
+            headers=[("Content-Type", "application/json"), auth_header],
+        )
+        resp_json = json.loads(response.get_data(as_text=True))
+        assert response.status_code == 400
+        assert resp_json["errors"][0]["message"] == expected_msg
+
+
 class TestSeedingBounceRateData:
     @freeze_time("2019-01-01 12:00:00.000000")
     @pytest.mark.parametrize(