diff --git a/nebula-logger/main/log-management/classes/LogBatchPurger.cls b/nebula-logger/main/log-management/classes/LogBatchPurger.cls
index 648df6fe6..5e2de0686 100644
--- a/nebula-logger/main/log-management/classes/LogBatchPurger.cls
+++ b/nebula-logger/main/log-management/classes/LogBatchPurger.cls
@@ -9,12 +9,65 @@
  * @see LogBatchPurgeScheduler
  */
 global with sharing class LogBatchPurger implements Database.Batchable<SObject>, Database.Stateful {
+    private final Boolean isSystemDebuggingEnabled;
+
     private String originalTransactionId;
     private Integer totalProcessedRecords = 0;
 
+    @testVisible
+    // Database.emptyRecycleBin counts as a DML statement per record
+    private static Integer MAX_RECORDS_DELETED = (Limits.getLimitDmlRows() / 2) - 1;
+    private static Integer DELETED_COUNT = 0;
+
     private class LogBatchPurgerException extends Exception {
     }
 
+    global LogBatchPurger() {
+        this.isSystemDebuggingEnabled = Logger.getUserSettings()?.EnableSystemMessages__c == true;
+    }
+
+    private class LogDeleter implements System.Queueable {
+        private final List<SObject> recordsToDelete;
+        public LogDeleter(List<SObject> recordsToDelete) {
+            this.recordsToDelete = recordsToDelete;
+        }
+
+        public void process() {
+            if (DELETED_COUNT + this.recordsToDelete.size() < MAX_RECORDS_DELETED) {
+                this.hardDelete(this.recordsToDelete);
+                this.recordsToDelete.clear();
+            } else {
+                List<SObject> safeToDeleteRecords = new List<SObject>();
+                while (this.recordsToDelete.size() > MAX_RECORDS_DELETED && !this.recordsToDelete.isEmpty()) {
+                    for (Integer index = this.recordsToDelete.size() - 1; index >= 0; index--) {
+                        safeToDeleteRecords.add(this.recordsToDelete[index]);
+                        this.recordsToDelete.remove(index);
+                    }
+                }
+                this.hardDelete(safeToDeleteRecords);
+            }
+
+            if (!this.recordsToDelete.isEmpty() && Limits.getLimitQueueableJobs() > Limits.getQueueableJobs()) {
+                System.enqueueJob(this);
+            }
+        }
+
+        public void execute(QueueableContext queueableContext) {
+            this.process();
+        }
+
+        private void hardDelete(List<SObject> records) {
+            // normally this would be an anti-pattern since most DML operations
+            // are a no-op with an empty list - but emptyRecycleBin throws
+            // for empty lists!
+            if (!records.isEmpty()) {
+                DELETED_COUNT += records.size();
+                delete records;
+                Database.emptyRecycleBin(records);
+            }
+        }
+    }
+
     global Database.QueryLocator start(Database.BatchableContext batchableContext) {
         if (!Schema.Log__c.SObjectType.getDescribe().isDeletable()) {
             throw new LogBatchPurgerException('User does not have access to delete logs');
@@ -24,7 +77,7 @@ global with sharing class LogBatchPurger implements Database.Batchable<SObject>,
         // ...so store the first transaction ID to later relate the other transactions
         this.originalTransactionId = Logger.getTransactionId();
 
-        if (Logger.getUserSettings().EnableSystemMessages__c == true) {
+        if (this.isSystemDebuggingEnabled) {
             Logger.info('Starting LogBatchPurger job');
             Logger.saveLog();
         }
@@ -39,24 +92,20 @@ global with sharing class LogBatchPurger implements Database.Batchable<SObject>,
             throw new LogBatchPurgerException('User does not have access to delete logs');
         }
 
-        this.totalProcessedRecords += logsToDelete.size();
-
         try {
-            if (Logger.getUserSettings().EnableSystemMessages__c == true) {
-                Logger.setParentLogTransactionId(this.originalTransactionId);
-                Logger.info(new LogMessage('Starting deletion of {0} records', logsToDelete.size()));
-            }
-
             // Delete the child log entries first
             List<LogEntry__c> logEntriesToDelete = [SELECT Id FROM LogEntry__c WHERE Log__c IN :logsToDelete];
-            delete logEntriesToDelete;
-            Database.emptyRecycleBin(logEntriesToDelete);
+            if (this.isSystemDebuggingEnabled) {
+                Logger.setParentLogTransactionId(this.originalTransactionId);
+                Logger.info(new LogMessage('Starting deletion of {0} logs and {1} log entries', logsToDelete.size(), logEntriesToDelete.size()));
+            }
+            new LogDeleter(logEntriesToDelete).process();
 
             // Now delete the parent logs
-            delete logsToDelete;
-            Database.emptyRecycleBin(logsToDelete);
+            new LogDeleter(logsToDelete).process();
+            this.totalProcessedRecords += DELETED_COUNT;
         } catch (Exception apexException) {
-            if (Logger.getUserSettings().EnableSystemMessages__c == true) {
+            if (this.isSystemDebuggingEnabled) {
                 Logger.error('Error deleting logs', apexException);
             }
         } finally {
@@ -65,7 +114,7 @@ global with sharing class LogBatchPurger implements Database.Batchable<SObject>,
     }
 
     global void finish(Database.BatchableContext batchableContext) {
-        if (Logger.getUserSettings().EnableSystemMessages__c == true) {
+        if (this.isSystemDebuggingEnabled) {
             Logger.setParentLogTransactionId(this.originalTransactionId);
             Logger.info(new LogMessage('Finished LogBatchPurger job, {0} total log records processed', this.totalProcessedRecords));
             Logger.saveLog();
diff --git a/nebula-logger/tests/log-management/classes/LogBatchPurger_Tests.cls b/nebula-logger/tests/log-management/classes/LogBatchPurger_Tests.cls
index f78b589f3..a86f64bb7 100644
--- a/nebula-logger/tests/log-management/classes/LogBatchPurger_Tests.cls
+++ b/nebula-logger/tests/log-management/classes/LogBatchPurger_Tests.cls
@@ -92,6 +92,21 @@ private class LogBatchPurger_Tests {
         System.assertEquals(0, logEntries.size(), logEntries);
     }
 
+    @isTest
+    static void it_should_continue_deleting_logs_even_when_current_count_greater_than_limits() {
+        List<Log__c> logsToDelete = [SELECT Id FROM Log__c];
+        LogBatchPurger.MAX_RECORDS_DELETED = logsToDelete.size() / 2;
+
+        Test.startTest();
+        new LogBatchPurger().execute(null, logsToDelete);
+        Test.stopTest();
+
+        logsToDelete = [SELECT Id FROM Log__c];
+        List<LogEntry__c>  logEntries = [SELECT Id FROM LogEntry__c];
+        System.assertEquals(0, logsToDelete.size(), logsToDelete);
+        System.assertEquals(0, logEntries.size(), logEntries);
+    }
+
     @isTest
     static void it_should_not_delete_a_log_before_scheduled_deletion_date() {
         List<Log__c> logs = [SELECT Id, LogRetentionDate__c FROM Log__c];