diff --git a/octodns/manager.py b/octodns/manager.py
index d710e91..c49c716 100644
--- a/octodns/manager.py
+++ b/octodns/manager.py
@@ -10,6 +10,7 @@ from importlib.metadata import PackageNotFoundError
 from importlib.metadata import version as module_version
 from json import dumps
 from logging import getLogger
+from re import compile as re_compile
 from sys import stdout
 
 from . import __version__
@@ -591,30 +592,49 @@ class Manager(object):
         the call and the zones returned from this function should be used
         instead.
         '''
-        for name, config in list(zones.items()):
-            if not name.startswith('*'):
+
+        # sorting longest first with the assumption that'll longer wildcards or
+        # regexes will be more specific, but mostly it's just to make the
+        # behavior consistent
+        for name, config in sorted(
+            zones.items(), key=lambda d: len(d[0]), reverse=True
+        ):
+            if name[0] != '*' and name[-1] != '$':
+                # this isn't a dynamic zone config, move along
                 continue
-            # we've found a dynamic config element
 
-            # find its sources
+            # it's dynamic, get a list of zone names from the configured sources
             found_sources = sources or self._get_sources(
                 name, config, eligible_sources
             )
             self.log.info('sync:   dynamic zone=%s, sources=%s', name, sources)
+            sourced_zones = set()
             for source in found_sources:
                 if not hasattr(source, 'list_zones'):
                     raise ManagerException(
                         f'dynamic zone={name} includes a source, {source.id}, that does not support `list_zones`'
                     )
-                for zone_name in source.list_zones():
-                    if zone_name in zones:
-                        self.log.info(
-                            'sync:      zone=%s already in config, ignoring',
-                            zone_name,
-                        )
-                        continue
-                    self.log.info('sync:     adding dynamic zone=%s', zone_name)
-                    zones[zone_name] = config
+                sourced_zones |= set(source.list_zones())
+
+            self.log.debug('_preprocess_zones: sourced_zones=%s', sourced_zones)
+
+            if name[-1] == '$':
+                # it's an end-anchored regex
+                re = re_compile(name)
+                # filter the zones we sourced with it
+                sourced_zones = set(z for z in sourced_zones if re.match(z))
+            # old-style wildcards are implcit catch-alls so they don't need
+            # filtering
+
+            # we do want to remove any explicitly configured zones or those
+            # that matched a previous wildcard/regex
+            sourced_zones -= set(zones.keys())
+
+            self.log.debug('_preprocess_zones: filtered=%s', sourced_zones)
+
+            for match in sourced_zones:
+                self.log.info('sync:     adding dynamic zone=%s', match)
+                zones[match] = config
 
             # remove the dynamic config element so we don't try and populate it
             del zones[name]
diff --git a/tests/test_octodns_manager.py b/tests/test_octodns_manager.py
index 7d18c7b..549658d 100644
--- a/tests/test_octodns_manager.py
+++ b/tests/test_octodns_manager.py
@@ -1394,6 +1394,39 @@ class TestManager(TestCase):
         )
         mock_source.list_zones.assert_called_once()
 
+        # doesn't matter what the actual name is, just that it starts with a *,
+        mock_source.reset_mock()
+        config = {'foo': 42}
+        zones = {'*SDFLKJSDFL': config, 'two': {'bar': 43}}
+        mock_source.list_zones.return_value = ['one', 'two', 'three']
+        got = manager._preprocess_zones(zones, sources=[mock_source])
+        self.assertEqual(
+            {'one': config, 'two': {'bar': 43}, 'three': config}, got
+        )
+        mock_source.list_zones.assert_called_once()
+
+        # multiple wildcards, this didn't make sense previously as the 2nd one
+        # would just win
+        mock_source.reset_mock()
+        config_a = {'foo': 42}
+        config_b = {'bar': 43}
+        zones = {r'.*\.a\.com\.$': config_a, r'.*\.b\.com\.$': config_b}
+        mock_source.list_zones.return_value = [
+            'one.a.com.',
+            'two.a.com.',
+            'three.b.com.',
+        ]
+        got = manager._preprocess_zones(zones, sources=[mock_source])
+        self.assertEqual(
+            {
+                'one.a.com.': config_a,
+                'two.a.com.': config_a,
+                'three.b.com.': config_b,
+            },
+            got,
+        )
+        self.assertEqual(2, mock_source.list_zones.call_count)
+
 
 class TestMainThreadExecutor(TestCase):
     def test_success(self):