diff --git a/.github/workflows/plugin-ci-workflow.yml b/.github/workflows/plugin-ci-workflow.yml
index b61f8775..49b07e66 100644
--- a/.github/workflows/plugin-ci-workflow.yml
+++ b/.github/workflows/plugin-ci-workflow.yml
@@ -214,6 +214,7 @@ jobs:
       run: |
         php tests/regression/threshold_recovery_database_test.php
         php tests/regression/quality_must_fix_database_test.php
+        php tests/regression/incoming_preservation_database_test.php
       working-directory: ${{ github.workspace }}/cacti/plugins/syslog
       env:
         SYSLOG_TEST_MYSQL_DSN: 'mysql:host=127.0.0.1;port=3306'
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b681bef4..0baf7537 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,14 @@
 
 --- develop ---
 
+* issue#317: Expire partitions by their actual UTC upper boundaries instead of partition count, preserving cutoff-spanning partitions and supporting legacy TO_DAYS layouts.
+
+* issue#317: Defer automatic partition reorganization when overflow estimates exceed 100,000 rows or 128 MiB, or cannot be read; preserve the safety partition and report the reason through maintenance health.
+
+* issue#317: Make incoming transfers transactional, lock overlapping slices, and skip existing source identities on retry to prevent duplicate local archival.
+
+* issue#317: Preserve unarchived incoming messages regardless of age, retry crash-stranded rows, and require a matching history or durable outbox identity before removing incoming data.
+
 * style: Keep all SonarCloud-reported lines within the 120-character limit by reflowing long SQL calls, ternaries, mappings and misaligned PHPDoc comment blocks, and associate the Device Alert Rule filter controls (Search, Enabled, Rules) with accessible labels
 * bug: Suppress automatic schema repair during plugin lifecycle operations and while Syslog is disabled or uninstalled; stop stale processor launches before they can recreate removed tables
 
diff --git a/README.md b/README.md
index 7cdc8675..26c90d54 100644
--- a/README.md
+++ b/README.md
@@ -161,8 +161,43 @@ Cacti core release.
   not grant rule-management permissions to untrusted users, and review custom
   SQL expressions before enabling them.
 
+### Incoming Transfer Recovery
+
+Incoming transfers lock their processing slice and commit history, incoming
+acknowledgement, and any remote outbox writes in one transaction. With InnoDB
+history, a failed or disconnected transfer rolls back and can be retried.
+Retries skip history identities already stored for the same collector, incoming
+sequence, and event timestamp, while separate incoming events remain separate.
+
+Aria history does not participate in transaction rollback. The identity check
+prevents duplicate history on a transfer retry, but full atomic recovery across
+a database or host crash requires InnoDB. Unarchived incoming rows are retained
+regardless of their event age, including rows missing host or program references.
+
 ### Partition Maintenance
 
+Partition retention uses actual upper boundaries against a UTC calendar cutoff,
+with today counted as one retained day. A seven-day setting retains today plus
+the previous six UTC dates, independently of partition count or the future
+pre-create window. Fully expired partitions are dropped even after gaps in
+maintenance. A partition spanning the cutoff is kept whole to protect newer
+messages, so some older messages may remain until that partition fully expires.
+Unknown boundaries or partition expressions stop pruning safely; indefinite
+retention and the `dMaxValue` safety partition remain preserved.
+
+
+Before each automatic reorganization, maintenance checks `dMaxValue` partition
+statistics under its creation lock. Recovery stops if estimated occupancy exceeds
+100,000 rows or 128 MiB of data plus indexes, or if size metadata is unavailable.
+The Status tab and pipeline health report the deferral and request operator
+intervention. The safety partition remains writable and retention pruning for
+the affected table stays deferred. An operator must plan recovery outside the
+poller; automatic recovery resumes when the overflow falls within these limits.
+These checks contain large recovery operations but do not impose an ALTER
+execution-time limit: statistics are estimates and concurrent writes can grow
+overflow after the check.
+
+
 Partition boundaries are calculated with integer UTC epoch arithmetic. When a
 new partition cannot be created safely, maintenance leaves the `dMaxValue`
 partition in place and skips retention pruning rather than risking a write gap.
diff --git a/includes/functions.php b/includes/functions.php
index fb7a6636..bbbba4ce 100644
--- a/includes/functions.php
+++ b/includes/functions.php
@@ -2360,8 +2360,8 @@ function syslog_partition_recover($table, $base_time, $ahead_days) {
 
 		$missing = $need_ahead - $created;
 
-		if (!syslog_partition_create($table, $time)) {
-			$stop = __('partition creation failed', 'syslog');
+		if (!syslog_partition_create($table, $time, $create_reason)) {
+			$stop = $create_reason !== '' ? $create_reason : __('partition creation failed', 'syslog');
 			/*
 			 * When the next concrete partition is still missing, every new
 			 * write lands in dMaxValue until the gap is healed; the
@@ -2725,14 +2725,17 @@ function syslog_partition_report_state($table) {
 /**
  * Create a new partition for the specified table.
  *
- * @param mixed $table The table to rotate
- * @param int   $time  Assume this time for the partition rotation
+ * @param mixed       $table  The table to rotate.
+ * @param int         $time   Assume this time for the partition rotation.
+ * @param string|null $reason Receives an overflow deferral reason, otherwise an empty string.
  *
  * @return bool true on success, false on lock failure or disallowed table.
  */
-function syslog_partition_create($table, $time = null) {
+function syslog_partition_create($table, $time = null, &$reason = null) {
 	global $syslogdb_default;
 
+	$reason = '';
+
 	if (!syslog_partition_table_allowed($table)) {
 		return false;
 	}
@@ -2817,6 +2820,34 @@ function syslog_partition_create($table, $time = null) {
 			[$syslogdb_default, $cformat, $table]);
 
 		if (!cacti_sizeof($exists)) {
+			$overflow = syslog_db_fetch_row_prepared('SELECT TABLE_ROWS AS overflow_rows,
+				DATA_LENGTH + INDEX_LENGTH AS overflow_bytes
+				FROM information_schema.PARTITIONS
+				WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? AND PARTITION_NAME = ?',
+				[$syslogdb_default, $table, 'dMaxValue']);
+
+			if (!isset($overflow['overflow_rows'], $overflow['overflow_bytes'])
+				|| !is_numeric($overflow['overflow_rows']) || !is_numeric($overflow['overflow_bytes'])
+				|| $overflow['overflow_rows'] < 0 || $overflow['overflow_bytes'] < 0) {
+				$reason = __(
+					'Automatic partition recovery deferred: dMaxValue size is unavailable; operator intervention required.',
+					'syslog'
+				);
+			} elseif ($overflow['overflow_rows'] > 100000 || $overflow['overflow_bytes'] > 134217728) {
+				$reason = __(
+					'Automatic partition recovery deferred: dMaxValue has an estimated %s rows and %s bytes; '
+					. 'limits are 100000 rows and 134217728 bytes. Operator intervention required.',
+					$overflow['overflow_rows'], $overflow['overflow_bytes'], 'syslog'
+				);
+			}
+
+			// shortcut: estimates and concurrent writes cannot bound ALTER duration; use staged recovery for strict bounds.
+			if ($reason !== '') {
+				cacti_log("SYSLOG WARNING: $table: $reason", false, 'SYSLOG');
+
+				return false;
+			}
+
 			cacti_log("SYSLOG: Creating new partition '$cformat'", false, 'SYSTEM');
 
 			syslog_debug("Creating new partition '$cformat'");
@@ -2916,63 +2947,74 @@ function syslog_partition_remove($table) {
 	$syslog_deleted = 0;
 
 	try {
-		$number_of_partitions = syslog_db_fetch_assoc_prepared('SELECT *
+		$partitions = syslog_db_fetch_assoc_prepared('SELECT PARTITION_NAME, PARTITION_DESCRIPTION, PARTITION_EXPRESSION
 			FROM `information_schema`.`partitions`
 			WHERE table_schema = ? AND table_name = ?
 			ORDER BY partition_ordinal_position',
 			[$syslogdb_default, $table]);
 
-		$days       = read_config_option('syslog_retention');
-		$ahead_days = syslog_partition_ahead_days();
+		$days = read_config_option('syslog_retention');
+
+		if (!is_array($partitions) || !ctype_digit((string) $days) || (int) $days < 1) {
+			return 0;
+		}
 
-		syslog_debug("There are currently '" . sizeof($number_of_partitions) . "' Syslog Partitions, We will keep '$days' retention partition(s) plus '$ahead_days' future partition(s).");
+		// Retention counts today as one UTC calendar day, matching new-table creation.
+		$cutoff = max(0, intdiv(time(), 86400) - ((int) $days - 1)) * 86400;
+		$day_cutoff = null;
 
-		if ($days > 0) {
-			$user_partitions = sizeof($number_of_partitions) - 1;
-			$keep_partitions = (int) $days + $ahead_days;
+		foreach ($partitions as $partition) {
+			$part_name = $partition['PARTITION_NAME'] ?? '';
+			$boundary  = $partition['PARTITION_DESCRIPTION'] ?? '';
+			$expression = strtolower(str_replace(['`', ' '], '', $partition['PARTITION_EXPRESSION'] ?? ''));
 
-			if ($user_partitions >= $keep_partitions) {
-				$i = 0;
+			if ($part_name === 'dMaxValue' || $boundary === 'MAXVALUE') {
+				break;
+			}
 
-				while ($user_partitions > $keep_partitions) {
-					$oldest = $number_of_partitions[$i];
+			if (preg_match('/^[a-zA-Z0-9_]+$/', $part_name) !== 1
+				|| !ctype_digit((string) $boundary)
+				|| !in_array($expression, ['unix_timestamp(logtime)', 'to_days(logtime)'], true)) {
+				cacti_log("SYSLOG ERROR: Invalid partition metadata for '$part_name' on '$table'; stopping retention prune", false, 'SYSLOG');
 
-					$part_name = $oldest['PARTITION_NAME'];
+				break;
+			}
 
-					if (preg_match('/^[a-zA-Z0-9_]+$/', $part_name) !== 1) {
-						cacti_log("SYSLOG ERROR: Invalid partition name '$part_name' for '$table'; skipping drop", false, 'SYSLOG');
+			$expiry = $cutoff;
 
-						break;
-					}
+			if ($expression === 'to_days(logtime)') {
+				if ($day_cutoff === null) {
+					$day_cutoff = syslog_db_fetch_cell_prepared('SELECT TO_DAYS(?)', [gmdate('Y-m-d', $cutoff)]);
+				}
 
-					if ($part_name === 'dMaxValue') {
-						// The dMaxValue partition is the write-path fail
-						// safe and must never be dropped.
-						cacti_log("SYSLOG ERROR: Refusing to drop the dMaxValue safety partition on '$table'; stopping retention prune", false, 'SYSLOG');
+				if (!ctype_digit((string) $day_cutoff)) {
+					cacti_log("SYSLOG ERROR: Unable to compute UTC day cutoff for '$table'; stopping retention prune", false, 'SYSLOG');
 
-						break;
-					}
+					break;
+				}
 
-					cacti_log("SYSLOG: Removing old partition '" . $part_name . "'", false, 'SYSTEM');
+				$expiry = (int) $day_cutoff;
+			}
 
-					syslog_debug("Removing partition '" . $part_name . "'");
+			// The upper boundary is exclusive: keep any partition spanning the cutoff.
+			if ((int) $boundary > $expiry) {
+				break;
+			}
 
-					// $table passed syslog_partition_table_allowed() at function entry;
-					// $part_name is regex-validated above. DDL identifiers cannot be parameterized.
-					$result = syslog_db_execute_prepared("ALTER TABLE `$syslogdb_default`.`$table` DROP PARTITION `$part_name`", []);
+			cacti_log("SYSLOG: Removing expired partition '$part_name'", false, 'SYSTEM');
 
-					if ($result === false) {
-						cacti_log("SYSLOG ERROR: Failed to drop partition '$part_name' from '$table' after $i successful drop(s); aborting further drops", false, 'SYSLOG');
+			// Both identifiers are validated; DDL identifiers cannot be parameterized.
+			$result = syslog_db_execute_prepared("ALTER TABLE `$syslogdb_default`.`$table` DROP PARTITION `$part_name`", []);
 
-						break;
-					}
+			if ($result === false) {
+				cacti_log("SYSLOG ERROR: Failed to drop partition '$part_name' from '$table'; aborting further drops", false, 'SYSLOG');
 
-					$i++;
-					$user_partitions--;
-					$syslog_deleted++;
-				}
+				break;
 			}
+
+			$syslog_deleted++;
 		}
+
 	} finally {
 		syslog_db_fetch_cell_prepared('SELECT RELEASE_LOCK(?)', [$lock_name]);
 	}
@@ -5984,7 +6026,9 @@ function syslog_get_alert_sql(&$alert, $max_seq, $processing_boundary = true, $c
 function syslog_preprocess_incoming_records(): array {
 	global $syslogdb_default;
 
-	$max_seq = syslog_db_fetch_cell("SELECT MAX(seq) FROM `$syslogdb_default`.`syslog_incoming` WHERE status = 0");
+	$max_seq = syslog_db_fetch_cell("SELECT MAX(seq)
+		FROM `$syslogdb_default`.`syslog_incoming`
+		WHERE status IN (0, 1)");
 
 	if ($max_seq > 0) {
 		// flag all records with the status = 1 prior to moving
@@ -6367,25 +6411,6 @@ function syslog_normalize_reference_tables($max_seq) {
 		[$max_seq]);
 }
 
-/**
- * syslog_delete_stale_incoming - Delete records left in the incoming table
- * that are older than one hour.  These are records whose owning poller
- * crashed before they were transferred.
- *
- * @return int|false The number of stale records deleted
- */
-function syslog_delete_stale_incoming() {
-	global $syslogdb_default, $syslog_cnn;
-
-	syslog_db_execute("DELETE FROM `$syslogdb_default`.`syslog_incoming` WHERE logtime < DATE_SUB(NOW(), INTERVAL 1 HOUR)");
-
-	$stale = db_affected_rows($syslog_cnn);
-
-	syslog_debug(sprintf('Deleted %5s - Stale Message(s) from incoming', $stale));
-
-	return $stale;
-}
-
 /**
  * Whether this process is a configured Syslog Remote Poller.  Main and
  * ordinary remote collectors never create an outbox backlog.
@@ -7042,8 +7067,8 @@ function syslog_replication_deliver_online(): int {
  *
  * When seq bounds are supplied, only records within the inclusive seq
  * slice are transferred which allows parallel workers to operate on
- * disjoint slices.  The stale record deletion is intentionally left to
- * the caller in that case.
+ * disjoint slices. Incoming rows are deleted only after their source
+ * identity is present in history or the durable replication outbox.
  *
  * @param int $max_seq   The max_seq for rows in the syslog table
  * @param int $seq_start Optional first seq of a slice (0 for no bound)
@@ -7062,17 +7087,24 @@ function syslog_incoming_to_syslog($max_seq, $seq_start = 0, $seq_end = 0) {
 		$slice_param = [$seq_start, $seq_end];
 	}
 
-	$replication_transaction = false;
+	if (!syslog_db_execute('START TRANSACTION')) {
+		cacti_log('SYSLOG ERROR: Unable to start incoming transfer transaction', false, 'SYSLOG');
 
-	if (syslog_replication_is_enabled()) {
-		if (!syslog_db_execute('START TRANSACTION')) {
-			cacti_log('SYSLOG ERROR: Unable to start transaction for replication outbox', false, 'SYSLOG');
+		return ['moved' => 0, 'stale' => 0, 'success' => false];
+	}
 
-			return ['moved' => 0, 'stale' => 0, 'success' => false];
-		}
+	// Lock the incoming slice so overlapping retries cannot archive it twice.
+	if (!syslog_db_execute_prepared("UPDATE `$syslogdb_default`.`syslog_incoming` AS si
+		SET status = 1
+		WHERE si.status = 1 AND si.seq <= ? $slice_where", array_merge([$max_seq], $slice_param))) {
+		syslog_db_execute('ROLLBACK');
+		cacti_log('SYSLOG ERROR: Unable to lock incoming transfer rows', false, 'SYSLOG');
 
-		$replication_transaction = true;
-		$replication_where       = 'WHERE si.`status` = 1 AND si.`seq` <= ?' . $slice_where;
+		return ['moved' => 0, 'stale' => 0, 'success' => false];
+	}
+
+	if (syslog_replication_is_enabled()) {
+		$replication_where = 'WHERE si.`status` = 1 AND si.`seq` <= ?' . $slice_where;
 
 		if (!syslog_replication_enqueue_incoming($replication_where, array_merge([$max_seq], $slice_param), 'syslog')) {
 			syslog_db_execute('ROLLBACK');
@@ -7091,24 +7123,28 @@ function syslog_incoming_to_syslog($max_seq, $seq_start = 0, $seq_end = 0) {
 			(logtime, priority_id, facility_id, program_id, host_id, message, message_digest, threshold_breached, replication_source_poller_id, replication_source_event_id)
 			SELECT logtime, priority_id, facility_id, program_id, host_id, message, MD5(message), threshold_breached, replication_source_poller_id, replication_source_event_id
 			FROM (
-				SELECT logtime, priority_id, facility_id, sp.program_id, sh.host_id, message, si.threshold_breached,
+				SELECT si.logtime, si.priority_id, si.facility_id, sp.program_id, sh.host_id,
+					si.message, si.threshold_breached,
 					? AS replication_source_poller_id, si.seq AS replication_source_event_id
-				FROM syslog_incoming AS si
-				INNER JOIN syslog_hosts AS sh
+				FROM `$syslogdb_default`.`syslog_incoming` AS si
+				INNER JOIN `$syslogdb_default`.`syslog_hosts` AS sh
 				ON sh.host = si.host
-				INNER JOIN syslog_programs AS sp
+				INNER JOIN `$syslogdb_default`.`syslog_programs` AS sp
 				ON sp.program = si.program
-				WHERE si.`status` = 1
+				LEFT JOIN `$syslogdb_default`.`syslog` AS archived
+				ON archived.replication_source_poller_id = ?
+				AND archived.replication_source_event_id = si.seq
+				AND archived.logtime = si.logtime
+				WHERE archived.seq IS NULL
+				AND si.`status` = 1
 				AND si.`seq` <= ?
 				$slice_where
 			) AS merge",
-			array_merge([(int) $config['poller_id'], $max_seq], $slice_param));
+			array_merge([(int) $config['poller_id'], (int) $config['poller_id'], $max_seq], $slice_param));
 	}
 
 	if (!$archived) {
-		if ($replication_transaction) {
-			syslog_db_execute('ROLLBACK');
-		}
+		syslog_db_execute('ROLLBACK');
 
 		cacti_log('SYSLOG ERROR: Unable to archive incoming records', false, 'SYSLOG');
 
@@ -7124,46 +7160,38 @@ function syslog_incoming_to_syslog($max_seq, $seq_start = 0, $seq_end = 0) {
 
 	syslog_debug(sprintf('Moved   %5s - Message(s) to the syslog table', $moved));
 
-	if ($seq_start > 0 && $seq_end > 0) {
-		$deleted = syslog_db_execute_prepared("DELETE FROM `$syslogdb_default`.`syslog_incoming`
-			WHERE `status` = 1
-			AND `seq` BETWEEN ? AND ?",
-			[$seq_start, $seq_end]);
-	} else {
-		$deleted = syslog_db_execute_prepared("DELETE FROM `$syslogdb_default`.`syslog_incoming`
-			WHERE `status` = 1
-			AND `seq` <= ?",
-			[$max_seq]);
-	}
+	// A successful INSERT can skip rows whose reference joins did not match.
+	$receipt_table  = syslog_replication_is_enabled() ? 'syslog_replication_output' : 'syslog';
+	$receipt_poller = syslog_replication_is_enabled() ? 'source_poller_id' : 'replication_source_poller_id';
+	$receipt_event  = syslog_replication_is_enabled() ? 'source_event_id' : 'replication_source_event_id';
+	$deleted = syslog_db_execute_prepared("DELETE FROM `$syslogdb_default`.`syslog_incoming`
+		WHERE `status` = 1
+		AND `seq` <= ?" . ($seq_start > 0 && $seq_end > 0 ? ' AND `seq` BETWEEN ? AND ?' : '') . "
+		AND EXISTS (
+			SELECT 1 FROM `$syslogdb_default`.`$receipt_table` AS receipt
+			WHERE receipt.`$receipt_poller` = ?
+			AND receipt.`$receipt_event` = syslog_incoming.seq
+			AND receipt.logtime = syslog_incoming.logtime
+		)", array_merge([$max_seq], $slice_param, [(int) $config['poller_id']]));
 
 	syslog_debug(sprintf('Deleted %5s - Already Processed Message(s) from incoming', db_affected_rows($syslog_cnn)));
 
 	if (!$deleted) {
-		if ($replication_transaction) {
-			syslog_db_execute('ROLLBACK');
-		}
+		syslog_db_execute('ROLLBACK');
 
 		cacti_log('SYSLOG ERROR: Unable to delete archived incoming records', false, 'SYSLOG');
 
 		return ['moved' => 0, 'stale' => 0, 'success' => false];
 	}
 
-	if ($replication_transaction && !syslog_db_execute('COMMIT')) {
+	if (!syslog_db_execute('COMMIT')) {
 		syslog_db_execute('ROLLBACK');
-		cacti_log('SYSLOG ERROR: Unable to commit transfer and replication outbox transaction', false, 'SYSLOG');
+		cacti_log('SYSLOG ERROR: Unable to confirm incoming transfer commit; retry is safe', false, 'SYSLOG');
 
 		return ['moved' => 0, 'stale' => 0, 'success' => false];
 	}
 
-	if ($seq_start > 0 && $seq_end > 0) {
-		// The stale record cleanup is owned by the master after all
-		// transfer workers complete to avoid interleaved DELETEs.
-		$stale = 0;
-	} else {
-		$stale = syslog_delete_stale_incoming();
-	}
-
-	return ['moved' => $moved, 'stale' => $stale, 'success' => true];
+	return ['moved' => $moved, 'stale' => 0, 'success' => true];
 }
 
 /**
diff --git a/includes/schema.php b/includes/schema.php
index 3517533f..a415f6a6 100644
--- a/includes/schema.php
+++ b/includes/schema.php
@@ -418,10 +418,8 @@ function syslog_create_partitioned_syslog_table($engine = 'InnoDB', $days = 30,
 	 *
 	 * $days counts today as one of the retained days, so the historical
 	 * loop bound is $days - 1 (today plus $days - 1 prior days = $days
-	 * total). This must stay in lockstep with the $days + $ahead_days
-	 * "keep_partitions" math in syslog_partition_remove(), or newly
-	 * installed tables over-provision by one partition and the very next
-	 * retention prune immediately deletes the oldest one.
+	 * total). This matches the UTC calendar cutoff in syslog_partition_remove(),
+	 * so the initial oldest partition still contains retained events.
 	 *
 	 * $days = 0 is a valid "Indefinite" retention setting (never pruned by
 	 * syslog_partition_remove(), which only prunes when $days > 0), but
diff --git a/locales/po/cacti.pot b/locales/po/cacti.pot
index 8b65e0c2..917946ea 100644
--- a/locales/po/cacti.pot
+++ b/locales/po/cacti.pot
@@ -3509,3 +3509,12 @@ msgstr ""
 #: syslog_saved_searches.php
 msgid "Choose a field, operator, and value. Use AND, OR, NOT, or Group to build the query. Changes apply to this shared template for all users."
 msgstr ""
+
+#: includes/functions.php
+msgid "Automatic partition recovery deferred: dMaxValue size is unavailable; operator intervention required."
+msgstr ""
+
+#: includes/functions.php
+#, php-format
+msgid "Automatic partition recovery deferred: dMaxValue has an estimated %s rows and %s bytes; limits are 100000 rows and 134217728 bytes. Operator intervention required."
+msgstr ""
diff --git a/syslog_process.php b/syslog_process.php
index 3259ed9b..78f60d2a 100644
--- a/syslog_process.php
+++ b/syslog_process.php
@@ -377,8 +377,7 @@
 api_plugin_hook('plugin_syslog_after_processing');
 
 /**
- * move records from incoming to syslog table and remove
- * any stale records to to a poller crash
+ * move records from incoming to syslog table, retaining unarchived rows
  *
  * In parallel mode the transfer is spread across workers
  * over disjoint seq slices.  Slices are recomputed after
@@ -407,8 +406,8 @@
 		$launched = syslog_launch_workers('transfer', $transfer_slices, $run_id, $debug);
 
 		/*
-		 * The stale record cleanup, the stats aggregation and the
-		 * master's own exit all assume the transfer is complete, so
+		 * The stats aggregation and the master's own exit assume the transfer
+		 * is complete, so
 		 * block until every child has recorded its statistics and
 		 * unregistered itself.
 		 */
@@ -418,10 +417,9 @@
 
 		$moved = $worker_stats['moved'];
 
-		if ($worker_stats['success']) {
-			$stale = syslog_delete_stale_incoming();
-		} else {
-			$stale = 0;
+		$stale = 0;
+
+		if (!$worker_stats['success']) {
 			cacti_log('SYSLOG ERROR: One or more transfer workers failed; stale incoming records were retained', false, 'SYSLOG');
 		}
 
diff --git a/tests/Security/PartitionTableLockingTest.php b/tests/Security/PartitionTableLockingTest.php
index 86fd6cfa..82b9a829 100644
--- a/tests/Security/PartitionTableLockingTest.php
+++ b/tests/Security/PartitionTableLockingTest.php
@@ -20,13 +20,13 @@
 it('keeps syslog partition table locking and DDL identifiers safe', function () {
 	$functions = plugin_test_read_source('includes/functions.php');
 
-	// All five information_schema queries must be prepared statements
+	// All six information_schema queries must be prepared statements
 	// scoped to the requested table via a placeholder. Match only calls
 	// whose first argument contains 'information_schema' to exclude the
 	// GET_LOCK / RELEASE_LOCK uses of syslog_db_fetch_cell_prepared.
 	$partition_query_count = preg_match_all('/syslog_db_fetch_(?:row|assoc|cell)_prepared\s*\([^)]*information_schema/', $functions);
 
-	if ($partition_query_count === false || $partition_query_count !== 5) {
+	if ($partition_query_count === false || $partition_query_count !== 6) {
 		throw new RuntimeException('Partition queries are not consistently scoped to the requested table.');
 	}
 
diff --git a/tests/Unit/ArchiveStatementBalanceTest.php b/tests/Unit/ArchiveStatementBalanceTest.php
index e62177c9..b567f10e 100644
--- a/tests/Unit/ArchiveStatementBalanceTest.php
+++ b/tests/Unit/ArchiveStatementBalanceTest.php
@@ -36,12 +36,6 @@
 			return true;
 		}
 
-		// The unprepared stale-cleanup DELETE carries no placeholders,
-		// so it cannot trip HY093; anything else is unexpected here.
-		if ($sql === "DELETE FROM `syslog`.`syslog_incoming` WHERE logtime < DATE_SUB(NOW(), INTERVAL 1 HOUR)") {
-			return true;
-		}
-
 		throw new ExpectationFailedException("Unexpected syslog_db_execute() query: $sql");
 	});
 
@@ -52,12 +46,12 @@
 	});
 
 	// No placeholder may arrive without a bound parameter: the archival
-	// INSERT pairs one bound poller_id with the max_seq bound, while the
+	// INSERT binds poller_id for both the new row and the retry check, while the
 	// digest is computed in SQL and never bound.
 	expect(syslog_incoming_to_syslog(100)['success'])->toBeTrue()
 		->and(syslog_incoming_to_syslog(100, 51, 100)['success'])->toBeTrue();
 
-	expect($statements)->toHaveCount(4);
+	expect($statements)->toHaveCount(6);
 
 	foreach ($statements as [$sql, $params]) {
 		$tokens = syslog_test_count_placeholders($sql);
@@ -67,17 +61,17 @@
 	}
 
 	// Both archival variants must keep an unbound MD5(message) digest and
-	// exactly one bound poller_id; the sliced variant adds one bound
+	// two bound poller_id values; the sliced variant adds one bound
 	// seq pair.
-	[$archive_sql, $archive_params] = $statements[0];
-	$slice_sql = $statements[2][0];
+	[$archive_sql, $archive_params] = $statements[1];
+	$slice_sql = $statements[4][0];
 
 	expect($archive_sql)->toContain('INSERT INTO `syslog`.`syslog`')
 		->toContain('message_digest')
 		->toContain('MD5(message)');
-	expect(substr_count($archive_sql, '?'))->toBe(2)
-		->and($archive_params)->toBe([1, 100])
-		->and(substr_count($slice_sql, '?'))->toBe(4);
+	expect(substr_count($archive_sql, '?'))->toBe(3)
+		->and($archive_params)->toBe([1, 1, 100])
+		->and(substr_count($slice_sql, '?'))->toBe(5);
 });
 
 /**
diff --git a/tests/Unit/IncomingPreservationTest.php b/tests/Unit/IncomingPreservationTest.php
new file mode 100644
index 00000000..467aec4d
--- /dev/null
+++ b/tests/Unit/IncomingPreservationTest.php
@@ -0,0 +1,106 @@
+<?php
+
+it('requires a matching durable source identity before deleting incoming rows', function () {
+	syslog_load_plugin_source('includes/functions.php');
+	$GLOBALS['config']['poller_id'] = 1;
+	$GLOBALS['syslogdb_default'] = 'syslog';
+	$deletes = [];
+	test_override('syslog_db_execute_prepared', function ($sql, $params) use (&$deletes) {
+		if (str_starts_with($sql, 'DELETE FROM')) {
+			$deletes[] = [$sql, $params];
+		}
+
+		return true;
+	});
+	test_override('syslog_db_execute', function ($sql) {
+		expect($sql)->not->toContain('DELETE');
+
+		return true;
+	});
+
+	expect(syslog_incoming_to_syslog(100)['stale'])->toBe(0);
+	expect(syslog_incoming_to_syslog(100, 51, 100)['success'])->toBeTrue();
+	expect($deletes)->toHaveCount(2);
+	foreach ($deletes as [$sql, $params]) {
+		expect($sql)->toContain('AND EXISTS (')
+			->toContain('`syslog`.`syslog` AS receipt')
+			->toContain('receipt.`replication_source_poller_id` = ?')
+			->toContain('receipt.`replication_source_event_id` = syslog_incoming.seq')
+			->toContain('receipt.logtime = syslog_incoming.logtime')
+			->not->toContain('DATE_SUB');
+	}
+	expect($deletes[0][1])->toBe([100, 1])
+		->and($deletes[1][1])->toBe([100, 51, 100, 1]);
+});
+
+it('uses durable outbox identities for remote incoming deletion', function () {
+	syslog_load_plugin_source('includes/functions.php');
+	$GLOBALS['config']['poller_id'] = 2;
+	$GLOBALS['syslogdb_default'] = 'syslog';
+	test_override('read_config_option', fn($name) => in_array($name, ['syslog_remote_enabled', 'syslog_remote_store_records'], true) ? 'on' : '');
+	$delete = '';
+	test_override('syslog_db_execute_prepared', function ($sql) use (&$delete) {
+		if (str_starts_with($sql, 'DELETE FROM')) {
+			$delete = $sql;
+		}
+
+		return true;
+	});
+
+	expect(syslog_incoming_to_syslog(100)['success'])->toBeTrue();
+	expect($delete)->toContain('`syslog`.`syslog_replication_output` AS receipt')
+		->toContain('receipt.`source_poller_id` = ?')
+		->toContain('receipt.`source_event_id` = syslog_incoming.seq');
+});
+
+
+it('retries crash-stranded rows even when no new messages arrive', function () {
+	syslog_load_plugin_source('includes/functions.php');
+	$GLOBALS['syslogdb_default'] = 'syslog';
+	test_override('syslog_db_fetch_cell', function ($sql) {
+		expect($sql)->toContain('WHERE status IN (0, 1)');
+
+		return 42;
+	});
+	test_override('syslog_db_execute_prepared', function ($sql, $params) {
+		expect($sql)->toContain('SET `status` = 1')->toContain('WHERE `status` = 0');
+		expect($params)->toBe([42]);
+
+		return true;
+	});
+	test_override('syslog_db_fetch_cell_prepared', static fn() => 3);
+	expect(syslog_preprocess_incoming_records())->toBe(['max_seq' => 42, 'incoming' => 3]);
+});
+
+it('rolls back failed local transfers without reporting archived rows', function ($failure) {
+	syslog_load_plugin_source('includes/functions.php');
+	$GLOBALS['config']['poller_id'] = 1;
+	$GLOBALS['syslogdb_default'] = 'syslog';
+	$calls = [];
+	test_override('syslog_db_execute', function ($sql) use (&$calls, $failure) {
+		$calls[] = $sql;
+
+		return $sql !== $failure;
+	});
+	test_override('syslog_db_execute_prepared', function ($sql) use (&$calls, $failure) {
+		$operation = strtok($sql, ' ');
+		$calls[] = $operation;
+
+		return $operation !== $failure;
+	});
+	test_override('db_affected_rows', static fn() => 4);
+
+	expect(syslog_incoming_to_syslog(100, 51, 100))->toBe(['moved' => 0, 'stale' => 0, 'success' => false]);
+	expect($calls[0])->toBe('START TRANSACTION');
+	if ($failure === 'START TRANSACTION') {
+		expect($calls)->toHaveCount(1);
+	} else {
+		expect(end($calls))->toBe('ROLLBACK');
+	}
+	if ($failure === 'UPDATE') {
+		expect($calls)->not->toContain('INSERT')->not->toContain('DELETE');
+	}
+	if ($failure === 'INSERT') {
+		expect($calls)->not->toContain('DELETE');
+	}
+})->with(['START TRANSACTION', 'UPDATE', 'INSERT', 'DELETE', 'COMMIT']);
diff --git a/tests/Unit/PartitionAheadTest.php b/tests/Unit/PartitionAheadTest.php
index 35968920..4ca4d794 100644
--- a/tests/Unit/PartitionAheadTest.php
+++ b/tests/Unit/PartitionAheadTest.php
@@ -59,6 +59,10 @@
 	});
 
 	test_override('syslog_db_fetch_row_prepared', function ($sql, $params = []) {
+		if (str_contains($sql, 'AS overflow_rows')) {
+			return ['overflow_rows' => 0, 'overflow_bytes' => 0];
+		}
+
 		if (str_contains($sql, 'information_schema')) {
 			return [];
 		}
@@ -117,16 +121,18 @@
 	});
 
 	test_override('syslog_db_fetch_assoc_prepared', function () {
-		return [
-			['PARTITION_NAME' => 'd20260910'],
-			['PARTITION_NAME' => 'd20260911'],
-			['PARTITION_NAME' => 'd20260912'],
-			['PARTITION_NAME' => 'd20260913'],
-			['PARTITION_NAME' => 'd20260914'],
-			['PARTITION_NAME' => 'd20260915'],
-			['PARTITION_NAME' => 'd20260916'],
-			['PARTITION_NAME' => 'dMaxValue']
-		];
+		$today = intdiv(time(), 86400) * 86400;
+		$partitions = [];
+		for ($day = -3; $day <= 3; $day++) {
+			$partitions[] = [
+				'PARTITION_NAME' => 'd' . gmdate('Ymd', $today + $day * 86400),
+				'PARTITION_DESCRIPTION' => (string) ($today + ($day + 1) * 86400),
+				'PARTITION_EXPRESSION' => 'unix_timestamp(`logtime`)',
+			];
+		}
+		$partitions[] = ['PARTITION_NAME' => 'dMaxValue'];
+
+		return $partitions;
 	});
 
 	test_override('syslog_db_execute_prepared', function ($sql) use (&$dropped) {
@@ -138,5 +144,5 @@
 	});
 
 	expect(syslog_partition_remove('syslog'))->toBe(1);
-	expect($dropped)->toBe(['d20260910']);
+	expect($dropped)->toBe(['d' . gmdate('Ymd', time() - 3 * 86400)]);
 });
diff --git a/tests/Unit/PartitionFailSafeTest.php b/tests/Unit/PartitionFailSafeTest.php
index 8522fa55..7970117d 100644
--- a/tests/Unit/PartitionFailSafeTest.php
+++ b/tests/Unit/PartitionFailSafeTest.php
@@ -108,6 +108,10 @@ function partition_failsafe_healthy_state(): void {
 	});
 
 	test_override('syslog_db_fetch_row_prepared', function ($sql, $params = []) {
+		if (str_contains($sql, 'AS overflow_rows')) {
+			return ['overflow_rows' => 0, 'overflow_bytes' => 0];
+		}
+
 		if (str_contains($sql, 'information_schema')) {
 			return [];
 		}
@@ -178,6 +182,10 @@ function partition_failsafe_healthy_state(): void {
 	});
 
 	test_override('syslog_db_fetch_row_prepared', function ($sql, $params = []) {
+		if (str_contains($sql, 'AS overflow_rows')) {
+			return ['overflow_rows' => 0, 'overflow_bytes' => 0];
+		}
+
 		if (str_contains($sql, 'information_schema')) {
 			return [];
 		}
@@ -266,7 +274,7 @@ function partition_failsafe_healthy_state(): void {
 
 	expect(syslog_partition_manage())->toBeGreaterThanOrEqual(0);
 	expect($values['partition_maintenance_blocked'])->toBe('0');
-	expect($dropped)->toBe([], 'No retention drop is due when the partition count is within the keep window.');
+	expect($dropped)->toBe([], 'No retention drop is due when retention is unset.');
 });
 
 it('never removes the dMaxValue safety partition', function () {
diff --git a/tests/Unit/PartitionOverflowGuardTest.php b/tests/Unit/PartitionOverflowGuardTest.php
new file mode 100644
index 00000000..e1b1066b
--- /dev/null
+++ b/tests/Unit/PartitionOverflowGuardTest.php
@@ -0,0 +1,113 @@
+<?php
+
+it('checks overflow size under the creation lock and defers unsafe reorganizations', function ($metadata, $blocked) {
+	syslog_load_plugin_source('includes/functions.php');
+	$GLOBALS['syslogdb_default'] = 'syslog_test';
+
+	foreach (['syslog', 'syslog_removed'] as $table) {
+		$locked = false;
+		$released = 0;
+		$alters = [];
+		test_override('syslog_db_fetch_cell_prepared', function ($sql) use (&$locked, &$released) {
+			if (str_contains($sql, 'GET_LOCK')) {
+				$locked = true;
+
+				return 1;
+			}
+			if (str_contains($sql, 'RELEASE_LOCK')) {
+				$locked = false;
+				$released++;
+
+				return 1;
+			}
+
+			return 'd20260101';
+		});
+		test_override('syslog_db_fetch_row_prepared', function ($sql, $params) use ($metadata, &$locked, $table, $blocked) {
+			if (str_contains($sql, 'AS overflow_rows')) {
+				expect($locked)->toBeTrue();
+				expect($params)->toBe(['syslog_test', $table, 'dMaxValue']);
+
+				return $metadata;
+			}
+			if (str_contains($sql, 'SHOW CREATE TABLE')) {
+				expect($blocked)->toBeFalse('Blocked recovery must stop before inspecting DDL.');
+
+				return ['Create Table' => 'PARTITION BY RANGE (UNIX_TIMESTAMP(logtime))'];
+			}
+
+			return [];
+		});
+		test_override('syslog_db_execute_prepared', function ($sql) use (&$alters) {
+			$alters[] = $sql;
+
+			return true;
+		});
+
+		$result = syslog_partition_recover($table, 1767398400, 0);
+		expect($released)->toBe(1)->and($locked)->toBeFalse();
+		expect($result['retention_deferred'])->toBe($blocked);
+		if ($blocked) {
+			expect($result['created'])->toBe(0)->and($result['missing'])->toBe(1);
+			expect($result['stop_reason'])->toContain('dMaxValue')->toContain('intervention required');
+			expect($alters)->toBe([]);
+		} else {
+			expect($result['created'])->toBe(1)->and($result['stop_reason'])->toBe('');
+			expect($alters)->toHaveCount(1);
+			expect($alters[0])->toContain('REORGANIZE PARTITION dMaxValue');
+		}
+	}
+})->with([
+	'empty' => [['overflow_rows' => 0, 'overflow_bytes' => 0], false],
+	'at limits' => [['overflow_rows' => '100000', 'overflow_bytes' => '134217728'], false],
+	'row limit exceeded' => [['overflow_rows' => 100001, 'overflow_bytes' => 0], true],
+	'byte limit exceeded' => [['overflow_rows' => 0, 'overflow_bytes' => 134217729], true],
+	'missing partition metadata' => [[], true],
+	'failed metadata query' => [false, true],
+	'unknown rows' => [['overflow_rows' => null, 'overflow_bytes' => 0], true],
+	'unknown bytes' => [['overflow_rows' => 0, 'overflow_bytes' => null], true],
+	'invalid rows' => [['overflow_rows' => 'unknown', 'overflow_bytes' => 0], true],
+	'invalid bytes' => [['overflow_rows' => 0, 'overflow_bytes' => 'unknown'], true],
+	'negative rows' => [['overflow_rows' => -1, 'overflow_bytes' => 0], true],
+	'negative bytes' => [['overflow_rows' => 0, 'overflow_bytes' => -1], true],
+]);
+
+it('does not apply overflow limits when the requested partition already exists', function () {
+	syslog_load_plugin_source('includes/functions.php');
+	$GLOBALS['syslogdb_default'] = 'syslog_test';
+	test_override('syslog_db_fetch_cell_prepared', static fn() => 1);
+	test_override('syslog_db_fetch_row_prepared', function ($sql) {
+		expect($sql)->not->toContain('AS overflow_rows');
+
+		return ['partition_name' => 'd20260103'];
+	});
+	$reason = 'previous deferral';
+	expect(syslog_partition_create('syslog', 1767398400, $reason))->toBeTrue();
+	expect($reason)->toBe('');
+});
+
+it('persists the overflow deferral reason without pruning either affected table', function () {
+	syslog_load_plugin_source('includes/functions.php');
+	$GLOBALS['syslogdb_default'] = 'syslog_test';
+	$values = [];
+	test_override('syslog_db_table_exists', static fn() => true);
+	test_override('syslog_db_fetch_assoc_prepared', static fn() => [
+		['partition_name' => 'd20260101', 'partition_description' => '1767312000', 'partition_ordinal_position' => 1],
+		['partition_name' => 'dMaxValue', 'partition_description' => 'MAXVALUE', 'partition_ordinal_position' => 2],
+	]);
+	test_override('syslog_db_fetch_cell_prepared', static fn($sql) => str_contains($sql, 'PARTITION_NAME') ? 'd20260101' : 1);
+	test_override('syslog_db_fetch_row_prepared', static fn($sql) => str_contains($sql, 'AS overflow_rows')
+		? ['overflow_rows' => 100001, 'overflow_bytes' => 0] : []);
+	test_override('syslog_db_execute_prepared', function ($sql, $params) use (&$values) {
+		expect($sql)->not->toContain('REORGANIZE PARTITION')->not->toContain('DROP PARTITION');
+		if (str_contains($sql, 'syslog_status')) {
+			$values[$params[0]] = $params[1];
+		}
+
+		return true;
+	});
+
+	expect(syslog_partition_manage())->toBe(0);
+	expect($values['partition_maintenance_blocked'])->toBe('1');
+	expect($values['partition_maintenance_reason'])->toContain('100001 rows')->toContain('intervention required');
+});
diff --git a/tests/Unit/PartitionRetentionBoundaryTest.php b/tests/Unit/PartitionRetentionBoundaryTest.php
new file mode 100644
index 00000000..42dbafb8
--- /dev/null
+++ b/tests/Unit/PartitionRetentionBoundaryTest.php
@@ -0,0 +1,91 @@
+<?php
+
+it('expires partitions by UTC upper boundary regardless of names, gaps or timezone', function ($expression, $straddles) {
+	syslog_load_plugin_source('includes/functions.php');
+	$GLOBALS['syslogdb_default'] = 'syslog_test';
+	$original_timezone = date_default_timezone_get();
+	$cutoff = (intdiv(time(), 86400) - 6) * 86400;
+	$day_mode = $expression === 'TO_DAYS(`logtime`)';
+	$expiry = $day_mode ? 1000 : $cutoff;
+	$unit = $day_mode ? 1 : 86400;
+	$rows = [
+		['PARTITION_NAME' => 'misleading_future_label', 'PARTITION_DESCRIPTION' => (string) ($expiry - 5 * $unit)],
+		['PARTITION_NAME' => 'expires_at_cutoff', 'PARTITION_DESCRIPTION' => (string) ($straddles ? $expiry + $unit : $expiry)],
+		['PARTITION_NAME' => 'spans_cutoff', 'PARTITION_DESCRIPTION' => (string) ($expiry + 2 * $unit)],
+		['PARTITION_NAME' => 'future', 'PARTITION_DESCRIPTION' => (string) ($expiry + 10 * $unit)],
+		['PARTITION_NAME' => 'dMaxValue', 'PARTITION_DESCRIPTION' => 'MAXVALUE'],
+	];
+	foreach ($rows as &$row) { $row['PARTITION_EXPRESSION'] = $expression; }
+	unset($row);
+	test_override('read_config_option', static fn($name) => $name === 'syslog_retention' ? '7' : '');
+	test_override('syslog_db_fetch_assoc_prepared', function ($sql, $params) use ($rows) {
+		expect($params)->toBe(['syslog_test', 'syslog']);
+
+		return $rows;
+	});
+	test_override('syslog_db_fetch_cell_prepared', function ($sql, $params) use ($cutoff) {
+		if ($sql === 'SELECT TO_DAYS(?)') {
+			expect($params)->toBe([gmdate('Y-m-d', $cutoff)]);
+
+			return '1000';
+		}
+
+		return 1;
+	});
+	try {
+		foreach (['UTC', 'America/New_York', 'Pacific/Auckland'] as $timezone) {
+			date_default_timezone_set($timezone);
+			$dropped = [];
+			test_override('syslog_db_execute_prepared', function ($sql) use (&$dropped) {
+				preg_match('/DROP PARTITION `([^`]+)`/', $sql, $matches);
+				$dropped[] = $matches[1];
+
+				return true;
+			});
+			expect(syslog_partition_remove('syslog'))->toBe($straddles ? 1 : 2);
+			expect($dropped)->toBe($straddles ? ['misleading_future_label']
+				: ['misleading_future_label', 'expires_at_cutoff']);
+		}
+	} finally { date_default_timezone_set($original_timezone); }
+})->with([
+	['UNIX_TIMESTAMP(`logtime`)', false], ['UNIX_TIMESTAMP(`logtime`)', true],
+	['TO_DAYS(`logtime`)', false], ['TO_DAYS(`logtime`)', true],
+]);
+
+it('preserves indefinite retention and stops safely on invalid metadata or failed drops', function ($case) {
+	syslog_load_plugin_source('includes/functions.php');
+	$GLOBALS['syslogdb_default'] = 'syslog_test';
+	$row = [
+		'PARTITION_NAME' => 'expired',
+		'PARTITION_DESCRIPTION' => '1',
+		'PARTITION_EXPRESSION' => 'unix_timestamp(logtime)',
+	];
+	if ($case === 'bad name') { $row['PARTITION_NAME'] = 'unsafe`name'; }
+	if ($case === 'missing boundary') { unset($row['PARTITION_DESCRIPTION']); }
+	if ($case === 'bad boundary') { $row['PARTITION_DESCRIPTION'] = 'unknown'; }
+	if ($case === 'bad expression') { $row['PARTITION_EXPRESSION'] = 'unknown(logtime)'; }
+	if ($case === 'cutoff failure') { $row['PARTITION_EXPRESSION'] = 'to_days(logtime)'; }
+	if ($case === 'safety partition') { $row['PARTITION_NAME'] = 'dMaxValue'; }
+	if ($case === 'renamed safety partition') { $row['PARTITION_DESCRIPTION'] = 'MAXVALUE'; }
+	$days = match ($case) { 'indefinite' => '0', 'invalid retention' => 'oops', default => '1' };
+	$released = 0;
+	$attempts = 0;
+	test_override('read_config_option', static fn() => $days);
+	test_override('syslog_db_fetch_assoc_prepared', static fn() => $case === 'metadata failure' ? false : [$row, $row]);
+	test_override('syslog_db_fetch_cell_prepared', function ($sql) use (&$released) {
+		if (str_contains($sql, 'RELEASE_LOCK')) { $released++; }
+
+		return $sql === 'SELECT TO_DAYS(?)' ? false : 1;
+	});
+	test_override('syslog_db_execute_prepared', function () use (&$attempts) {
+		$attempts++;
+
+		return false;
+	});
+
+	expect(syslog_partition_remove('syslog_removed'))->toBe(0);
+	expect($released)->toBe(1)->and($attempts)->toBe($case === 'drop failure' ? 1 : 0);
+})->with([
+	'indefinite', 'invalid retention', 'metadata failure', 'bad name', 'missing boundary',
+	'bad boundary', 'bad expression', 'cutoff failure', 'safety partition', 'renamed safety partition', 'drop failure',
+]);
diff --git a/tests/Unit/ReplicationCoverageRegressionTest.php b/tests/Unit/ReplicationCoverageRegressionTest.php
index 907accc9..bc78a5f2 100644
--- a/tests/Unit/ReplicationCoverageRegressionTest.php
+++ b/tests/Unit/ReplicationCoverageRegressionTest.php
@@ -729,7 +729,7 @@ function cacti_escapeshellarg($value) {
 	test_override('read_config_option', fn () => '');
 	expect(syslog_incoming_to_syslog(100))->toBe([
 		'moved' => 3,
-		'stale' => 3,
+		'stale' => 0,
 		'success' => true,
 	]);
 });
diff --git a/tests/Unit/SyslogWorkerAggregationTest.php b/tests/Unit/SyslogWorkerAggregationTest.php
index d9657e94..1ed6d844 100644
--- a/tests/Unit/SyslogWorkerAggregationTest.php
+++ b/tests/Unit/SyslogWorkerAggregationTest.php
@@ -101,7 +101,7 @@
 
 	expect($process)->toContain('$success = $results[\'success\'];')
 		->toContain('\'success\'  => $success,')
-		->toContain('if ($worker_stats[\'success\'])')
+		->toContain('if (!$worker_stats[\'success\'])')
 		->toContain('stale incoming records were retained');
 });
 
diff --git a/tests/bootstrap-unit.php b/tests/bootstrap-unit.php
index 04b1be88..c6587b12 100644
--- a/tests/bootstrap-unit.php
+++ b/tests/bootstrap-unit.php
@@ -355,6 +355,12 @@ function api_plugin_register_realm($plugin, $files, $description, $login) {
 	}
 }
 
+if (!function_exists('api_plugin_hook')) {
+	function api_plugin_hook($name) {
+		return test_call_override('api_plugin_hook', [$name], null);
+	}
+}
+
 if (!function_exists('api_plugin_hook_function')) {
 	function api_plugin_hook_function($name, $value = null) {
 		return test_call_override('api_plugin_hook_function', [$name, $value], $value);
diff --git a/tests/regression/incoming_preservation_database_test.php b/tests/regression/incoming_preservation_database_test.php
new file mode 100644
index 00000000..702589b8
--- /dev/null
+++ b/tests/regression/incoming_preservation_database_test.php
@@ -0,0 +1,235 @@
+<?php
+// Run only against an explicitly supplied disposable MariaDB server.
+$socket = getenv('SYSLOG_TEST_MYSQL_SOCKET');
+$dsn = $socket ? 'mysql:unix_socket=' . $socket : getenv('SYSLOG_TEST_MYSQL_DSN');
+if (!$dsn) {
+	fwrite(STDERR, "Set SYSLOG_TEST_MYSQL_SOCKET or SYSLOG_TEST_MYSQL_DSN for a disposable test server.\n");
+	exit(2);
+}
+$pdo = new PDO($dsn, 'root', getenv('SYSLOG_TEST_MYSQL_PASSWORD') ?: '', [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]);
+$syslogdb_default = 'syslog_incoming_' . bin2hex(random_bytes(4));
+$config = ['poller_id' => 1];
+$options = [];
+$affected = 0;
+$checks = 0;
+function check_incoming($condition, $message) {
+	$GLOBALS['checks']++;
+	if (!$condition) { throw new RuntimeException($message); }
+}
+function syslog_db_execute_prepared($sql, $params = [], ...$args) {
+	if (!empty($GLOBALS['fail_sql']) && str_starts_with($sql, $GLOBALS['fail_sql'])) {
+		return false;
+	}
+	if (str_starts_with($sql, 'DELETE FROM') && isset($GLOBALS['before_delete'])) {
+		$callback = $GLOBALS['before_delete'];
+		unset($GLOBALS['before_delete']);
+		$callback();
+	}
+	$q = $GLOBALS['pdo']->prepare($sql);
+	try {
+		$success = $q->execute($params);
+	} catch (PDOException $error) {
+		if (($error->errorInfo[1] ?? 0) !== 1205) { throw $error; }
+		return false;
+	}
+	$GLOBALS['affected'] = $q->rowCount();
+	return $success;
+}
+function syslog_db_execute($sql, ...$args) {
+	$success = syslog_db_execute_prepared($sql);
+	return $sql === 'COMMIT' && !empty($GLOBALS['lose_commit_ack']) ? false : $success;
+}
+function syslog_db_fetch_cell_prepared($sql, $params = [], ...$args) {
+	$q = $GLOBALS['pdo']->prepare($sql);
+	$q->execute($params);
+	return $q->fetchColumn();
+}
+function syslog_db_fetch_cell($sql) { return syslog_db_fetch_cell_prepared($sql); }
+function syslog_db_fetch_assoc_prepared($sql, $params = [], ...$args) {
+	$q = $GLOBALS['pdo']->prepare($sql);
+	$q->execute($params);
+	return $q->fetchAll(PDO::FETCH_ASSOC);
+}
+function syslog_db_fetch_row_prepared($sql, $params = [], ...$args) {
+	$q = $GLOBALS['pdo']->prepare($sql);
+	$q->execute($params);
+	return $q->fetch(PDO::FETCH_ASSOC) ?: [];
+}
+function cacti_sizeof($items) { return is_array($items) ? count($items) : 0; }
+function __($text, ...$args) {
+	array_pop($args);
+	return $args ? vsprintf($text, $args) : $text;
+}
+function db_affected_rows(...$args) { return $GLOBALS['affected']; }
+function read_config_option($name) { return $GLOBALS['options'][$name] ?? ''; }
+function api_plugin_hook(...$args) {}
+function cacti_log(...$args) {}
+require dirname(__DIR__, 2) . '/includes/functions.php';
+function incoming_ids() {
+	return $GLOBALS['pdo']->query('SELECT seq FROM syslog_incoming ORDER BY seq')->fetchAll(PDO::FETCH_COLUMN);
+}
+try {
+	$pdo->exec("CREATE DATABASE `$syslogdb_default`");
+	$pdo->exec("USE `$syslogdb_default`");
+	$pdo->exec('CREATE TABLE syslog_incoming (
+		seq BIGINT PRIMARY KEY, status TINYINT DEFAULT 0, logtime TIMESTAMP NOT NULL,
+		host VARCHAR(64), program VARCHAR(40), message TEXT,
+		priority_id INT, facility_id INT, threshold_breached TINYINT) ENGINE=InnoDB');
+	$pdo->exec('CREATE TABLE syslog (
+		seq BIGINT AUTO_INCREMENT, logtime TIMESTAMP NOT NULL,
+		priority_id INT, facility_id INT, program_id INT, host_id INT,
+		message TEXT, message_digest CHAR(32), threshold_breached TINYINT,
+		replication_source_poller_id INT, replication_source_event_id BIGINT,
+		PRIMARY KEY(seq, logtime), KEY replication_source(replication_source_poller_id, replication_source_event_id)
+		) ENGINE=InnoDB PARTITION BY RANGE(UNIX_TIMESTAMP(logtime)) (PARTITION dMaxValue VALUES LESS THAN MAXVALUE)');
+	$pdo->exec('CREATE TABLE syslog_hosts (host_id INT PRIMARY KEY, host VARCHAR(64))');
+	$pdo->exec('CREATE TABLE syslog_programs (program_id INT PRIMARY KEY, program VARCHAR(40))');
+	$pdo->exec("INSERT INTO syslog_hosts VALUES (1, 'router')");
+	$pdo->exec("INSERT INTO syslog_programs VALUES (1, 'app')");
+	$pdo->exec("INSERT INTO syslog_incoming (seq,status,logtime,host,program,message) VALUES
+		(1,1,NOW() - INTERVAL 2 DAY,'router','app','crash-stranded'),
+		(2,1,NOW(),NULL,'app','missing host'),
+		(3,1,NOW(),'router',NULL,'missing program'),
+		(4,1,NOW(),'unknown','app','failed host normalization'),
+		(5,1,NOW(),'router','unknown','failed program normalization'),
+		(6,1,NOW(),'router','app','valid')");
+	$batch = syslog_preprocess_incoming_records();
+	check_incoming($batch === ['max_seq' => 6, 'incoming' => 6], 'An entirely crash-stranded queue must be retried');
+	$pdo->exec("INSERT INTO syslog_incoming (seq,logtime,host,program,message)
+		VALUES (7,NOW() - INTERVAL 2 DAY,'router','app','late arrival after snapshot')");
+	check_incoming(syslog_incoming_to_syslog(6)['success'], 'Archival succeeds with unmatched rows');
+	check_incoming(incoming_ids() === [2,3,4,5,7], 'NULLs, missing references and late arrivals must survive');
+	check_incoming((int)$pdo->query('SELECT COUNT(*) FROM syslog')->fetchColumn() === 2, 'Both valid messages are archived');
+	$pdo->exec("INSERT INTO syslog_hosts VALUES (2, 'unknown')");
+	$pdo->exec("INSERT INTO syslog_programs VALUES (2, 'unknown')");
+	$batch = syslog_preprocess_incoming_records();
+	check_incoming($batch['max_seq'] === 7, 'Retry includes retained rows and newly arrived rows');
+	check_incoming(syslog_incoming_to_syslog(7,4,5)['success'], 'Sliced retry succeeds');
+	check_incoming(incoming_ids() === [2,3,7], 'Only repaired rows in the worker slice are removed');
+	check_incoming(syslog_incoming_to_syslog(7,7,7)['success'], 'Delayed event is archived');
+	check_incoming(incoming_ids() === [2,3], 'Missing fields stay queued without age-based deletion');
+
+	// A failed delete must roll back both the history insert and incoming changes.
+	$pdo->exec("INSERT INTO syslog_incoming (seq,status,logtime,host,program,message)
+		VALUES (8,1,NOW(),'router','app','atomic transfer')");
+	$before = (int)$pdo->query('SELECT COUNT(*) FROM syslog')->fetchColumn();
+	$fail_sql = 'DELETE FROM';
+	check_incoming(!syslog_incoming_to_syslog(8,8,8)['success'], 'Failed deletion reports transfer failure');
+	check_incoming((int)$pdo->query('SELECT COUNT(*) FROM syslog')->fetchColumn() === $before, 'History insert rolls back');
+	check_incoming(in_array(8, incoming_ids(), true), 'Failed deletion preserves incoming');
+	$fail_sql = 'COMMIT';
+	check_incoming(!syslog_incoming_to_syslog(8,8,8)['success'], 'Failed commit reports transfer failure');
+	check_incoming((int)$pdo->query('SELECT COUNT(*) FROM syslog')->fetchColumn() === $before, 'Failed commit rolls back history');
+	$fail_sql = '';
+	$lose_commit_ack = true;
+	check_incoming(!syslog_incoming_to_syslog(8,8,8)['success'], 'Unconfirmed committed transfer reports failure');
+	$lose_commit_ack = false;
+	check_incoming(syslog_incoming_to_syslog(8,8,8)['moved'] === 0, 'Retry after lost commit acknowledgement adds no history');
+	check_incoming((int)$pdo->query('SELECT COUNT(*) FROM syslog')->fetchColumn() === $before + 1, 'Exactly one copy survives');
+
+	// A disconnected process loses its uncommitted insert and preserves incoming.
+	$pdo->exec("INSERT INTO syslog_incoming (seq,status,logtime,host,program,message)
+		VALUES (9,1,NOW(),'router','app','disconnect before delete')");
+	$before_delete = function () { throw new RuntimeException('simulated process exit'); };
+	try { syslog_incoming_to_syslog(9,9,9); } catch (RuntimeException $error) {
+		check_incoming($error->getMessage() === 'simulated process exit', 'Process stops after archival before deletion');
+	}
+	$pdo = new PDO($dsn, 'root', getenv('SYSLOG_TEST_MYSQL_PASSWORD') ?: '', [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]);
+	$pdo->exec("USE `$syslogdb_default`");
+	check_incoming(in_array(9, incoming_ids(), true), 'Disconnect rolls back incoming state');
+	check_incoming(syslog_incoming_to_syslog(9,9,9)['moved'] === 1, 'Retry archives the disconnected transfer once');
+
+	// An independent connection must not archive a slice while its owner holds it.
+	$pdo->exec("INSERT INTO syslog_incoming (seq,status,logtime,host,program,message)
+		VALUES (10,1,NOW(),'router','app','overlapping retry')");
+	$before_delete = function () use ($dsn, $syslogdb_default) {
+		$owner = $GLOBALS['pdo'];
+		$GLOBALS['pdo'] = new PDO($dsn, 'root', getenv('SYSLOG_TEST_MYSQL_PASSWORD') ?: '', [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]);
+		$GLOBALS['pdo']->exec("USE `$syslogdb_default`");
+		$GLOBALS['pdo']->exec('SET innodb_lock_wait_timeout = 1');
+		try {
+			check_incoming(!syslog_incoming_to_syslog(10,10,10)['success'], 'Overlapping retry cannot bypass incoming locks');
+		} finally { $GLOBALS['pdo'] = $owner; }
+	};
+	check_incoming(syslog_incoming_to_syslog(10,10,10)['success'], 'Original slice commits after overlap is rejected');
+	check_incoming(syslog_incoming_to_syslog(10,10,10)['moved'] === 0, 'Overlapping retry adds no duplicate after commit');
+
+	// Aria history cannot roll back: a retry must recognize the persisted insert.
+	$pdo->exec('ALTER TABLE syslog ENGINE=Aria');
+	$pdo->exec("INSERT INTO syslog_incoming (seq,status,logtime,host,program,message)
+		VALUES (11,1,NOW(),'router','app','nontransactional retry')");
+	$fail_sql = 'DELETE FROM';
+	check_incoming(!syslog_incoming_to_syslog(11,11,11)['success'], 'Aria transfer failure is reported');
+	check_incoming(in_array(11, incoming_ids(), true), 'Aria failure preserves incoming');
+	$fail_sql = '';
+	check_incoming(syslog_incoming_to_syslog(11,11,11)['moved'] === 0, 'Aria retry skips its existing history identity');
+	check_incoming(!in_array(11, incoming_ids(), true), 'Aria retry acknowledges existing history');
+	check_incoming((int)$pdo->query('SELECT COUNT(*) FROM syslog WHERE replication_source_event_id = 11')->fetchColumn() === 1,
+		'Aria retry leaves exactly one history row');
+	$pdo->exec('ALTER TABLE syslog ENGINE=InnoDB');
+
+	$pdo->exec("INSERT INTO syslog_incoming (seq,status,logtime,host,program,message) VALUES
+		(12,1,'2026-10-16 12:00:00','router','app','identical event text'),
+		(13,1,'2026-10-16 12:00:00','router','app','identical event text')");
+	check_incoming(syslog_incoming_to_syslog(13,12,13)['moved'] === 2,
+		'Distinct incoming sequences with identical content remain distinct events');
+
+	// Remote archival may omit NULL references, but the raw outbox must preserve them.
+	$pdo->exec('CREATE TABLE syslog_replication_output (
+		source_poller_id INT, source_event_id BIGINT, facility_id INT, priority_id INT,
+		program VARCHAR(40), logtime TIMESTAMP NOT NULL, host VARCHAR(64), message TEXT,
+		threshold_breached TINYINT, disposition VARCHAR(16), PRIMARY KEY(source_poller_id,source_event_id)) ENGINE=InnoDB');
+	$config['poller_id'] = 2;
+	$options = ['syslog_remote_enabled' => 'on', 'syslog_remote_store_records' => 'on'];
+	check_incoming(syslog_incoming_to_syslog(7)['success'], 'Remote transfer commits with raw outbox copies');
+	check_incoming(incoming_ids() === [], 'Incoming is acknowledged after outbox preservation');
+	check_incoming((int)$pdo->query('SELECT COUNT(*) FROM syslog_replication_output')->fetchColumn() === 2, 'Both malformed rows survive in the outbox');
+	// Verify the metadata guard against a populated physical overflow partition.
+	$pdo->exec('CREATE TABLE syslog_removed LIKE syslog');
+	$pdo->exec('ALTER TABLE syslog_removed ENGINE=Aria');
+	$reason = '';
+	check_incoming(syslog_partition_create('syslog_removed', time(), $reason), 'Small overflow can be reorganized');
+	$pdo->exec("INSERT INTO syslog_removed (logtime,message)
+		SELECT '2030-01-01 00:00:00','overflow' FROM seq_1_to_100001");
+	check_incoming(!syslog_partition_create('syslog_removed', time() + 86400, $reason), 'Oversized real overflow is deferred');
+	check_incoming(str_contains($reason, 'intervention required'), 'Deferral returns the operator-facing reason');
+	$pdo->exec("INSERT INTO syslog_removed (logtime,message) VALUES ('2030-01-01 00:00:00','still writable')");
+	check_incoming((int)$pdo->query('SELECT COUNT(*) FROM syslog_removed PARTITION (dMaxValue)')->fetchColumn() === 100002,
+		'Deferred overflow remains intact and writable');
+
+	$options = ['syslog_retention' => '7'];
+	$cutoff = (intdiv(time(), 86400) - 6) * 86400;
+	foreach (['UNIX_TIMESTAMP', 'TO_DAYS'] as $expression) {
+		foreach ([false, true] as $spans_cutoff) {
+			$pdo->exec('DROP TABLE syslog_removed');
+			$pdo->exec("SET time_zone = '+00:00'");
+			$epochs = [$cutoff - 2 * 86400, $cutoff - ($spans_cutoff ? 86400 : 0), $cutoff + 86400, $cutoff + 10 * 86400];
+			$boundaries = [];
+			foreach ($epochs as $epoch) {
+				$boundaries[] = $expression === 'TO_DAYS'
+					? (int)syslog_db_fetch_cell_prepared('SELECT TO_DAYS(?)', [gmdate('Y-m-d', $epoch)]) : $epoch;
+			}
+			$type = $expression === 'TO_DAYS' ? 'DATETIME' : 'TIMESTAMP';
+			$pdo->exec("CREATE TABLE syslog_removed (id INT, logtime $type NOT NULL, PRIMARY KEY(id,logtime))
+				ENGINE=InnoDB PARTITION BY RANGE ($expression(logtime)) (
+				PARTITION misleading_label VALUES LESS THAN ($boundaries[0]),
+				PARTITION expires VALUES LESS THAN ($boundaries[1]),
+				PARTITION spans VALUES LESS THAN ($boundaries[2]),
+				PARTITION future VALUES LESS THAN ($boundaries[3]),
+				PARTITION dMaxValue VALUES LESS THAN MAXVALUE)");
+			foreach ([-3 * 86400, -129600, -3600, 0, 3 * 86400] as $index => $offset) {
+				syslog_db_execute_prepared('INSERT INTO syslog_removed VALUES (?,?)',
+					[$index + 1, gmdate('Y-m-d H:i:s', $cutoff + $offset)]);
+			}
+			$pdo->exec("SET time_zone = '-07:00'");
+			check_incoming(syslog_partition_remove('syslog_removed') === 2, "$expression sparse history expires by age");
+			$ids = $pdo->query('SELECT id FROM syslog_removed ORDER BY id')->fetchAll(PDO::FETCH_COLUMN);
+			check_incoming($ids === ($spans_cutoff ? [3,4,5] : [4,5]),
+				"$expression preserves cutoff events, future events, and mixed-age partitions");
+		}
+	}
+
+	echo "PASS: $checks incoming preservation database checks\n";
+} finally {
+	$pdo->exec("DROP DATABASE IF EXISTS `$syslogdb_default`");
+}