diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2155ed8..a9e1f79 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,8 @@
   octodns.com.octodns.com
 * Fixed issues with handling of chunking large TXT values for providers that use
   the in-built `rrs` method
+* ExcludeRootNsChanges processor that will error (or warn) if plan includes a
+  change to root NS records
 
 ## v1.2.1 - 2023-09-29 - Now with fewer stale files
 
diff --git a/README.md b/README.md
index 0250890..4c5cac0 100644
--- a/README.md
+++ b/README.md
@@ -330,6 +330,7 @@ Similar to providers, but can only serve to populate records into a zone, cannot
 |--|--|
 | [AcmeMangingProcessor](/octodns/processor/acme.py) | Useful when processes external to octoDNS are managing acme challenge DNS records, e.g. LetsEncrypt |
 | [AutoArpa](/octodns/processor/arpa.py) | See [Automatic PTR generation](#automatic-ptr-generation) below |
+| [ExcludeRootNsChanges](/octodns/processor/filter.py) | Filter that errors or warns on planned root/APEX NS records changes. |
 | [IgnoreRootNsFilter](/octodns/processor/filter.py) | Filter that INGORES root/APEX NS records and prevents octoDNS from trying to manage them (where supported.) |
 | [MetaProcessor](/octodns/processor/meta.py) | Adds a special meta record with timing, UUID, providers, and/or version to aid in debugging and monitoring. |
 | [NameAllowlistFilter](/octodns/processor/filter.py) | Filter that ONLY manages records that match specified naming patterns, all others will be ignored |
diff --git a/octodns/processor/filter.py b/octodns/processor/filter.py
index 4723073..2de3b23 100644
--- a/octodns/processor/filter.py
+++ b/octodns/processor/filter.py
@@ -2,6 +2,7 @@
 #
 #
 
+from logging import getLogger
 from re import compile as re_compile
 
 from ..record.exception import ValidationError
@@ -218,6 +219,56 @@ class IgnoreRootNsFilter(BaseProcessor):
     process_target_zone = _process
 
 
+class ExcludeRootNsChanges(BaseProcessor):
+    '''Do not allow root NS record changes
+
+    Example usage:
+
+    processors:
+      exclude-root-ns-changes:
+        class: octodns.processor.filter.ExcludeRootNsChanges
+        # If true an a change for a root NS is seen an error will be thrown. If
+        # false a warning will be printed and the change will be removed from
+        # the plan.
+        # (default: true)
+        error: true
+
+    zones:
+      exxampled.com.:
+        sources:
+          - config
+        processors:
+          - exclude-root-ns-changes
+        targets:
+          - ns1
+    '''
+
+    def __init__(self, name, error=True):
+        self.log = getLogger(f'ExcludeRootNsChanges[{name}]')
+        super().__init__(name)
+        self.error = error
+
+    def process_plan(self, plan, sources, target):
+        if plan:
+            for change in list(plan.changes):
+                record = change.record
+                if record._type == 'NS' and record.name == '':
+                    self.log.warning(
+                        'root NS changes are disallowed, fqdn=%s', record.fqdn
+                    )
+                    if self.error:
+                        raise ValidationError(
+                            record.fqdn,
+                            ['root NS changes are disallowed'],
+                            record.context,
+                        )
+                    plan.changes.remove(change)
+
+            print(len(plan.changes))
+
+        return plan
+
+
 class ZoneNameFilter(BaseProcessor):
     '''Filter or error on record names that contain the zone name
 
diff --git a/tests/test_octodns_processor_filter.py b/tests/test_octodns_processor_filter.py
index 9857926..2d9b881 100644
--- a/tests/test_octodns_processor_filter.py
+++ b/tests/test_octodns_processor_filter.py
@@ -5,6 +5,7 @@
 from unittest import TestCase
 
 from octodns.processor.filter import (
+    ExcludeRootNsChanges,
     IgnoreRootNsFilter,
     NameAllowlistFilter,
     NameRejectlistFilter,
@@ -12,7 +13,8 @@ from octodns.processor.filter import (
     TypeRejectlistFilter,
     ZoneNameFilter,
 )
-from octodns.record import Record
+from octodns.provider.plan import Plan
+from octodns.record import Record, Update
 from octodns.record.exception import ValidationError
 from octodns.zone import Zone
 
@@ -184,6 +186,57 @@ class TestIgnoreRootNsFilter(TestCase):
         )
 
 
+class TestExcludeRootNsChanges(TestCase):
+    zone = Zone('unit.tests.', [])
+    root = Record.new(
+        zone, '', {'type': 'NS', 'ttl': 42, 'value': 'ns1.unit.tests.'}
+    )
+    zone.add_record(root)
+    not_root = Record.new(
+        zone, 'sub', {'type': 'NS', 'ttl': 43, 'value': 'ns2.unit.tests.'}
+    )
+    zone.add_record(not_root)
+    not_ns = Record.new(zone, '', {'type': 'A', 'ttl': 42, 'value': '3.4.5.6'})
+    zone.add_record(not_ns)
+    changes_with_root = [
+        Update(root, root),
+        Update(not_root, not_root),
+        Update(not_ns, not_ns),
+    ]
+    plan_with_root = Plan(zone, zone, changes_with_root, True)
+    changes_without_root = [Update(not_root, not_root), Update(not_ns, not_ns)]
+    plan_without_root = Plan(zone, zone, changes_without_root, True)
+
+    def test_no_plan(self):
+        proc = ExcludeRootNsChanges('exclude-root')
+        self.assertFalse(proc.process_plan(None, None, None))
+
+    def test_error(self):
+        proc = ExcludeRootNsChanges('exclude-root')
+
+        with self.assertRaises(ValidationError) as ctx:
+            proc.process_plan(self.plan_with_root, None, None)
+        self.assertEqual(
+            ['root NS changes are disallowed'], ctx.exception.reasons
+        )
+
+        self.assertEqual(
+            self.plan_without_root,
+            proc.process_plan(self.plan_without_root, None, None),
+        )
+
+    def test_warning(self):
+        proc = ExcludeRootNsChanges('exclude-root', error=False)
+
+        filtered_plan = proc.process_plan(self.plan_with_root, None, None)
+        self.assertEqual(self.plan_without_root.changes, filtered_plan.changes)
+
+        self.assertEqual(
+            self.plan_without_root,
+            proc.process_plan(self.plan_without_root, None, None),
+        )
+
+
 class TestZoneNameFilter(TestCase):
     def test_ends_with_zone(self):
         zone_name_filter = ZoneNameFilter('zone-name', error=False)