diff --git a/docs/cli.rst b/docs/cli.rst
index 46ad820..974fe68 100644
--- a/docs/cli.rst
+++ b/docs/cli.rst
@@ -1074,6 +1074,10 @@ The ``sqlite-utils enable-counts`` command can be used to configure these trigge
     # Configure triggers just for specific tables
     $ sqlite-utils enable-counts mydb.db table1 table2
 
+If the ``_counts`` table ever becomes out-of-sync with the actual table counts you can repair it using the ``reset-counts`` command::
+
+    $ sqlite-utils reset-counts mydb.db
+
 .. _cli_vacuum:
 
 Vacuum
diff --git a/docs/python-api.rst b/docs/python-api.rst
index 8f943f6..28adfba 100644
--- a/docs/python-api.rst
+++ b/docs/python-api.rst
@@ -1768,6 +1768,12 @@ If the property is ``True`` any calls to the ``table.count`` property will first
 
 Calling the ``.enable_counts()`` method on a database or table object will set ``use_counts_table`` to ``True`` for the lifetime of that database object.
 
+If the ``_counts`` table ever becomes out-of-sync with the actual table counts you can repair it using the ``.reset_counts()`` method:
+
+.. code-block:: python
+
+    db.reset_counts()
+
 Creating indexes
 ================
 
diff --git a/sqlite_utils/db.py b/sqlite_utils/db.py
index bc954e5..dc2a2e8 100644
--- a/sqlite_utils/db.py
+++ b/sqlite_utils/db.py
@@ -320,6 +320,17 @@ class Database:
         except OperationalError:
             return {}
 
+    def reset_counts(self):
+        tables = [table for table in self.tables if table.has_counts_triggers]
+        with self.conn:
+            self._ensure_counts_table()
+            counts_table = self[self._counts_table_name]
+            counts_table.delete_where()
+            counts_table.insert_all(
+                {"table": table.name, "count": table.execute_count()}
+                for table in tables
+            )
+
     def execute_returning_dicts(self, sql, params=None):
         cursor = self.execute(sql, params or tuple())
         keys = [d[0] for d in cursor.description]
diff --git a/tests/test_enable_counts.py b/tests/test_enable_counts.py
index 6fabd78..b70378e 100644
--- a/tests/test_enable_counts.py
+++ b/tests/test_enable_counts.py
@@ -64,8 +64,10 @@ def test_enable_counts_all_tables(fresh_db):
 def counts_db_path(tmpdir):
     path = str(tmpdir / "test.db")
     db = Database(path)
-    db["foo"].insert({"name": "Cleo"})
-    db["bar"].insert({"name": "Cleo"})
+    db["foo"].insert({"name": "bar"})
+    db["bar"].insert({"name": "bar"})
+    db["bar"].insert({"name": "bar"})
+    db["baz"].insert({"name": "bar"})
     return path
 
 
@@ -79,6 +81,8 @@ def counts_db_path(tmpdir):
                 "foo_counts_delete",
                 "bar_counts_insert",
                 "bar_counts_delete",
+                "baz_counts_insert",
+                "baz_counts_delete",
             ],
         ),
         (
@@ -121,11 +125,40 @@ def test_uses_counts_after_enable_counts(counts_db_path):
         ("select name from sqlite_master where type = 'view'", None),
         ("select name from sqlite_master where type = 'view'", None),
         ("select name from sqlite_master where type = 'view'", None),
+        ("select name from sqlite_master where type = 'view'", None),
         ("select sql from sqlite_master where name = ?", ("foo",)),
         ("SELECT quote(:value)", {"value": "foo"}),
         ("select sql from sqlite_master where name = ?", ("bar",)),
         ("SELECT quote(:value)", {"value": "bar"}),
+        ("select sql from sqlite_master where name = ?", ("baz",)),
+        ("SELECT quote(:value)", {"value": "baz"}),
         ("select sql from sqlite_master where name = ?", ("_counts",)),
         ("select name from sqlite_master where type = 'view'", None),
         ("select [table], count from _counts where [table] in (?)", ["foo"]),
     ]
+
+
+def test_reset_counts(counts_db_path):
+    db = Database(counts_db_path)
+    db["foo"].enable_counts()
+    db["bar"].enable_counts()
+    assert db.cached_counts() == {"foo": 1, "bar": 2}
+    # Corrupt the value
+    db["_counts"].update("foo", {"count": 3})
+    assert db.cached_counts() == {"foo": 3, "bar": 2}
+    assert db["foo"].count == 3
+    # Reset them
+    db.reset_counts()
+    assert db.cached_counts() == {"foo": 1, "bar": 2}
+    assert db["foo"].count == 1
+
+
+def test_reset_counts_cli(counts_db_path):
+    db = Database(counts_db_path)
+    db["foo"].enable_counts()
+    db["bar"].enable_counts()
+    assert db.cached_counts() == {"foo": 1, "bar": 2}
+    db["_counts"].update("foo", {"count": 3})
+    result = CliRunner().invoke(cli.cli, ["reset-counts", counts_db_path])
+    assert result.exit_code == 0
+    assert db.cached_counts() == {"foo": 1, "bar": 2}