diff --git a/app/celery/tasks.py b/app/celery/tasks.py
index 167f1669c..a7ca9665c 100644
--- a/app/celery/tasks.py
+++ b/app/celery/tasks.py
@@ -37,7 +37,7 @@ from app.models import (
     JOB_STATUS_IN_PROGRESS,
     JOB_STATUS_FINISHED,
     JOB_STATUS_READY_TO_SEND,
-    JOB_STATUS_SENT_TO_DVLA)
+    JOB_STATUS_SENT_TO_DVLA, JOB_STATUS_ERROR)
 from app.notifications.process_notifications import persist_notification
 from app.service.utils import service_allowed_to_send_to
 from app.statsd_decorators import statsd
@@ -308,6 +308,13 @@ def update_job_to_sent_to_dvla(self, job_id):
                             "Updated {} job to {}".format(updated_count, job_id, JOB_STATUS_SENT_TO_DVLA))
 
 
+@notify_celery.task(bind=True, name='update-letter-job-to-error')
+@statsd(namespace="tasks")
+def update_dvla_job_to_error(self, job_id):
+    dao_update_job_status(job_id, JOB_STATUS_ERROR)
+    current_app.logger.info("Updated {} job to {}".format(job_id, JOB_STATUS_ERROR))
+
+
 def create_dvla_file_contents(job_id):
     file_contents = '\n'.join(
         str(LetterDVLATemplate(
diff --git a/app/models.py b/app/models.py
index 8fd971fd6..7b1b62cc2 100644
--- a/app/models.py
+++ b/app/models.py
@@ -448,6 +448,7 @@ JOB_STATUS_SCHEDULED = 'scheduled'
 JOB_STATUS_CANCELLED = 'cancelled'
 JOB_STATUS_READY_TO_SEND = 'ready to send'
 JOB_STATUS_SENT_TO_DVLA = 'sent to dvla'
+JOB_STATUS_ERROR = 'error'
 JOB_STATUS_TYPES = [
     JOB_STATUS_PENDING,
     JOB_STATUS_IN_PROGRESS,
@@ -456,7 +457,8 @@ JOB_STATUS_TYPES = [
     JOB_STATUS_SCHEDULED,
     JOB_STATUS_CANCELLED,
     JOB_STATUS_READY_TO_SEND,
-    JOB_STATUS_SENT_TO_DVLA
+    JOB_STATUS_SENT_TO_DVLA,
+    JOB_STATUS_ERROR
 ]
 
 
diff --git a/migrations/versions/0071_add_job_error_state.py b/migrations/versions/0071_add_job_error_state.py
new file mode 100644
index 000000000..4b6eaf5b4
--- /dev/null
+++ b/migrations/versions/0071_add_job_error_state.py
@@ -0,0 +1,23 @@
+"""empty message
+
+Revision ID: 0071_add_job_error_state
+Revises: 0070_fix_notify_user_email
+Create Date: 2017-03-10 16:15:22.153948
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '0071_add_job_error_state'
+down_revision = '0070_fix_notify_user_email'
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+    op.execute("INSERT INTO JOB_STATUS VALUES('error')")
+
+
+def downgrade():
+    op.execute("UPDATE jobs SET job_status = 'finished' WHERE job_status = 'error'")
+    op.execute("DELETE FROM JOB_STATUS WHERE name = 'error'")
diff --git a/tests/app/celery/test_tasks.py b/tests/app/celery/test_tasks.py
index 604c046e6..0a960329b 100644
--- a/tests/app/celery/test_tasks.py
+++ b/tests/app/celery/test_tasks.py
@@ -12,7 +12,7 @@ from celery.exceptions import Retry
 from app import (encryption, DATETIME_FORMAT)
 from app.celery import provider_tasks
 from app.celery import tasks
-from app.celery.tasks import s3, build_dvla_file, create_dvla_file_contents
+from app.celery.tasks import s3, build_dvla_file, create_dvla_file_contents, update_dvla_job_to_error
 from app.celery.tasks import (
     process_job,
     process_row,
@@ -1059,3 +1059,16 @@ def test_update_job_to_sent_to_dvla(sample_letter_template, sample_letter_job):
     assert [(n.status == 'sending', n.sent_by == 'dvla') for n in updated_notifications]
 
     assert 'sent to dvla' == Job.query.filter_by(id=sample_letter_job.id).one().job_status
+
+
+def test_update_dvla_job_to_error(sample_letter_template, sample_letter_job):
+    create_notification(template=sample_letter_template, job=sample_letter_job)
+    create_notification(template=sample_letter_template, job=sample_letter_job)
+    update_dvla_job_to_error(job_id=sample_letter_job.id)
+
+    updated_notifications = Notification.query.all()
+    for n in updated_notifications:
+        assert n.status == 'created'
+        assert not n.sent_by
+
+    assert 'error' == Job.query.filter_by(id=sample_letter_job.id).one().job_status
diff --git a/tests/app/job/test_rest.py b/tests/app/job/test_rest.py
index 40e30d38e..05e9d00e5 100644
--- a/tests/app/job/test_rest.py
+++ b/tests/app/job/test_rest.py
@@ -673,7 +673,7 @@ def test_get_jobs_accepts_page_parameter(
 @pytest.mark.parametrize('statuses_filter, expected_statuses', [
     ('', JOB_STATUS_TYPES),
     ('pending', [JOB_STATUS_PENDING]),
-    ('pending, in progress, finished, sending limits exceeded, scheduled, cancelled, ready to send, sent to dvla',
+    ('pending, in progress, finished, sending limits exceeded, scheduled, cancelled, ready to send, sent to dvla, error',  # noqa
      JOB_STATUS_TYPES),
     # bad statuses are accepted, just return no data
     ('foo', [])
@@ -694,6 +694,7 @@ def test_get_jobs_can_filter_on_statuses(
     create_job(notify_db, notify_db_session, job_status='cancelled')
     create_job(notify_db, notify_db_session, job_status='ready to send')
     create_job(notify_db, notify_db_session, job_status='sent to dvla')
+    create_job(notify_db, notify_db_session, job_status='error')
 
     path = '/service/{}/job'.format(sample_service.id)
     response = client.get(