Keep transaction-log retention on a durable sequence floor

AGENTS.md

@@ -274,6 +274,14 @@ sufficient (env teardown does not honor tsfn acquire counts); see
    use the marked logical prefix and never expose the orphaned physical tail. The marker is not copied
    into backups: the copied prefix is already a clean canonical `.txnlog`.
    A known-zero-byte failure leaves the segment and its zero marker reusable.
+   Only the **active** segment may have a marker _created_ for it (a retired segment keeps the one it
+   already earned), and startup discovery cannot know which segment that is until the whole directory
+   has been scanned — directory iteration order is unspecified, so any segment can briefly hold the
+   highest sequence. `registerLogFile()` therefore registers without opening, and `load()`
+   marker-enables and opens the surviving current file after discovery. Promoting eagerly minted a
+   marker for a segment that is never appended to and, on Windows (where the handle is opened without
+   `FILE_SHARE_DELETE`), left every superseded segment undeletable by anything outside the process for
+   the life of the store.
    **The physical extent tracks `size` on POSIX only.** There the fd is `O_APPEND`,
    so writes go to physical EOF, not to `size`, and leaving orphaned bytes makes every later
    append land after a partial entry: a mid-file framing break that `recoverTail()` deliberately

README.md

@@ -1620,6 +1620,12 @@ const names = db.listLogs();
 ### `db.purgeLogs({ includeEntryCounts: true, ...options }): { path: string; entries: number }[]`
 
 Deletes transaction log files older than the `transactionLogRetention` (defaults to 3 days).
+Ordinary retention keeps the sequence file named by `txn.state` and every newer file as the live
+store's durable floor, or the highest sequence file when there is no persisted flush position. It
+removes only an eligible contiguous prefix below that floor. An idle store can therefore retain one
+file past the cutoff until a later write rotates and flushes it; disk retained past the cutoff is
+bounded by that store's `transactionLogMaxSize` (except when a single transaction exceeds the
+target). Use `destroy: true` only to remove the store itself.
 
 - `options: object`
   - `before?: number` Remove all transaction log files older than the specified timestamp.
@@ -1796,7 +1802,8 @@ stats.totals.transactionsWritten; // lifetime count of transactions written
 
 The `purge.retainedUnflushedFiles` gauge is useful for diagnosing why logs are not being cleaned
 up: a file can be older than the retention period but still retained because its transactions have
-not yet been flushed to RocksDB (purging it would be unsafe for crash recovery).
+not yet been flushed to RocksDB (purging it would be unsafe for crash recovery). The durable floor
+named by `txn.state` is not counted as purgeable even when it is old and fully flushed.
 
 ### Transaction Log Initialization
 

docs/stats.md

@@ -477,7 +477,8 @@ const stats: TransactionLogStats = log.getStats();
   - `offset: number` The byte offset within that file.
 - `purge: object`
   - `oldestFileAgeMs: number` Age in milliseconds of the oldest file on disk.
-  - `purgeableFiles: number` Number of files eligible for purge under the retention policy.
+  - `purgeableFiles: number` Number of files below the durable `txn.state` sequence floor that are
+    eligible for purge under the retention policy.
   - `retainedUnflushedFiles: number` Number of files past the retention threshold but retained
     because they are unflushed.
   - `lastPurgeMs: number` Timestamp of the last purge scan.

docs/transaction-log.md

@@ -198,7 +198,9 @@ await db.transaction((txn) => {
 - Log files are automatically rotated when either the index or data file reaches their configured
   maximum sizes
 - Rotation happens on the next write after the size limit is exceeded
-- Old log files can be automatically purged based on retention policy
+- Old log files can be automatically purged based on retention policy. The sequence file named by
+  `txn.state` and every newer file form the live store's durable floor, so an idle store can keep
+  one bounded file past the cutoff until a later write rotates and flushes it.
 
 ### Error Handling
 

src/binding/transaction_log/transaction_log.h

@@ -25,4 +25,4 @@ struct TransactionLog final {
 
 } // namespace rocksdb_js
 
-#endif
+#endif

src/binding/transaction_log/transaction_log_store.cpp

@@ -366,9 +366,10 @@ void TransactionLogStore::ensureExtent(const std::shared_ptr<TransactionLogFile>
 	}
 
 	// Never borrow the active segment's handle. Its lifecycle belongs to the
-	// write path — registerLogFile()/getLogFile() open it before the first
-	// append — so the only way it reads size 0 and closed is the window between
-	// getLogFile() creating it and the first append, where 0 is the truth. A
+	// write path — load()'s post-discovery activation and getLogFile() open it
+	// before the first append — so the only way it reads size 0 and closed is the
+	// window between getLogFile() creating it and the first append, where 0 is
+	// the truth. A
 	// close() here would drop a handle (and, on Windows, the mapping) the next
 	// append expects to still hold.
 	if (file->sequenceNumber == this->currentSequenceNumber.load(std::memory_order_relaxed)) {
@@ -556,6 +557,12 @@ void TransactionLogStore::collectStats(TransactionLogStoreStats& out) {
 	}
 
 	const bool retentionEnabled = this->retentionMs.count() > 0;
+	uint32_t durableFloorSequence = this->sequenceFiles.empty()
+		? 0
+		: this->sequenceFiles.rbegin()->first;
+	if (this->sequenceFiles.find(flushedPosition.logSequenceNumber) != this->sequenceFiles.end()) {
+		durableFloorSequence = flushedPosition.logSequenceNumber;
+	}
 
 	for (const auto& [seq, logFile] : this->sequenceFiles) {
 		// A registered-but-never-opened older file still reads 0 here (see
@@ -615,9 +622,9 @@ void TransactionLogStore::collectStats(TransactionLogStoreStats& out) {
 				// extent — would retain it. Gauge-only skew; nothing acts on it.
 				bool fullyFlushed = !(seq > flushedPosition.logSequenceNumber ||
 					(seq == flushedPosition.logSequenceNumber && fileSize > flushedPosition.positionInLogFile));
-				if (fullyFlushed) {
+				if (fullyFlushed && seq != durableFloorSequence) {
 					out.purgeableFiles++;
-				} else {
+				} else if (!fullyFlushed) {
 					out.retainedUnflushedFiles++;
 				}
 			}
@@ -649,10 +656,18 @@ void TransactionLogStore::doPurge(std::function<void(const std::filesystem::path
 
 	// collect sequence numbers to remove to avoid modifying map during iteration
 	std::vector<uint32_t> sequenceNumbersToRemove;
+	auto lastFlushedPosition = this->getLastFlushedPosition();
+	uint32_t durableFloorSequence = this->sequenceFiles.rbegin()->first;
+	if (this->sequenceFiles.find(lastFlushedPosition.logSequenceNumber) != this->sequenceFiles.end()) {
+		durableFloorSequence = lastFlushedPosition.logSequenceNumber;
+	}
 
 	for (const auto& entry : this->sequenceFiles) {
 		auto& sequenceNumber = entry.first;
 		auto& logFile = entry.second;
+		if (!all && sequenceNumber == durableFloorSequence) {
+			break;
+		}
 		bool shouldPurge = all;
 
 		if (!shouldPurge && (before > 0 || this->retentionMs.count() > 0)) {
@@ -669,30 +684,36 @@ void TransactionLogStore::doPurge(std::function<void(const std::filesystem::path
 					shouldPurge = fileAgeMs > this->retentionMs;
 				}
 			} catch (const std::filesystem::filesystem_error& e) {
-				// file was deleted or doesn't exist
-				DEBUG_LOG("%p TransactionLogStore::purge File no longer exists: %s\n", this, logFile->path.string().c_str());
-				continue;
+				if (e.code() == std::errc::no_such_file_or_directory) {
+					DEBUG_LOG("%p TransactionLogStore::purge Forgetting missing file: %s\n", this, logFile->path.string().c_str());
+					sequenceNumbersToRemove.push_back(sequenceNumber);
+					continue;
+				}
+				DEBUG_LOG("%p TransactionLogStore::purge Failed to get last write time for file %s: %s\n", this, logFile->path.string().c_str(), e.what());
+				break;
 			} catch (const std::exception& e) {
 				DEBUG_LOG("%p TransactionLogStore::purge Failed to get last write time for file %s: %s\n", this, logFile->path.string().c_str(), e.what());
-				continue;
+				break;
 			} catch (...) {
 				auto eptr = std::current_exception();
 				std::string errorMsg = getExceptionMessage(eptr);
 				DEBUG_LOG("%p TransactionLogStore::purge Unknown error getting last write time for file %s: %s\n",
 					this, logFile->path.string().c_str(), errorMsg.c_str());
-				continue;
+				break;
 			}
 		}
 
 		if (!shouldPurge) {
-			continue;
+			break;
 		}
 
 		// only purge files that are entirely before the last flushed position,
 		// guaranteeing all their transactions have been committed to RocksDB
-		auto lastFlushedPosition = this->getLastFlushedPosition();
 		if (sequenceNumber > lastFlushedPosition.logSequenceNumber) {
-			continue;
+			if (all) {
+				continue;
+			}
+			break;
 		}
 		if (sequenceNumber == lastFlushedPosition.logSequenceNumber) {
 			// The only comparison here that reads `size`, and the only file whose
@@ -705,13 +726,22 @@ void TransactionLogStore::doPurge(std::function<void(const std::filesystem::path
 				// Unresolvable extent — refuse to purge rather than delete a
 				// segment we cannot prove is flushed.
 				DEBUG_LOG("%p TransactionLogStore::purge Failed to resolve extent for file %s: %s\n", this, logFile->path.string().c_str(), e.what());
-				continue;
+				if (all) {
+					continue;
+				}
+				break;
 			} catch (...) {
 				DEBUG_LOG("%p TransactionLogStore::purge Failed to resolve extent for file %s\n", this, logFile->path.string().c_str());
-				continue;
+				if (all) {
+					continue;
+				}
+				break;
 			}
 			if (logFile->size > lastFlushedPosition.positionInLogFile) {
-				continue;
+				if (all) {
+					continue;
+				}
+				break;
 			}
 		}
 
@@ -723,7 +753,10 @@ void TransactionLogStore::doPurge(std::function<void(const std::filesystem::path
 		uint32_t removedSize = logFile->size.load(std::memory_order_relaxed);
 		auto removed = logFile->removeFile();
 		if (!removed) {
-			continue;
+			if (all) {
+				continue;
+			}
+			break;
 		}
 		this->filesPurged.fetch_add(1, std::memory_order_relaxed);
 		this->bytesPurged.fetch_add(removedSize, std::memory_order_relaxed);
@@ -759,7 +792,7 @@ void TransactionLogStore::doPurge(std::function<void(const std::filesystem::path
 
 	// if all log files have been removed, clean up the empty directory
 	// only try to remove if we actually removed at least one file from this store
-	if (this->sequenceFiles.empty() && !sequenceNumbersToRemove.empty()) {
+	if (all && this->sequenceFiles.empty() && !sequenceNumbersToRemove.empty()) {
 		try {
 			if (std::filesystem::exists(this->path)) {
 				DEBUG_LOG("%p TransactionLogStore::purge Removing log store directory: %s\n", this, this->path.string().c_str());
@@ -813,12 +846,15 @@ void TransactionLogStore::registerLogFile(const std::filesystem::path& path, con
 
 	if (retiredBoundary == 0 &&
 		sequenceNumber >= this->currentSequenceNumber.load(std::memory_order_relaxed)) {
-		logFile->appendBoundaryMarkerEnabled = true;
-		if (!logFile->isOpen()) {
-			logFile->open(this->latestTimestamp);
-		}
+		// Record the candidate only; load() opens and marker-enables whichever file
+		// is still current once the whole directory has been scanned. Directory
+		// iteration order is unspecified, so any segment can hold the highest-seen
+		// sequence for a moment, and a segment opened on that transient promotion is
+		// never closed again: it keeps an append-boundary marker it will never use
+		// and, on Windows (no FILE_SHARE_DELETE), stays undeletable by anything
+		// outside this process for the life of the store.
 		this->currentSequenceNumber.store(sequenceNumber, std::memory_order_relaxed);
-		this->nextLogPosition = { logFile->size, sequenceNumber };
+		this->nextLogPosition = { 0, sequenceNumber };
 	} else if (retiredBoundary > 0 &&
 		sequenceNumber >= this->currentSequenceNumber.load(std::memory_order_relaxed)) {
 		uint32_t nextWritableSequence = sequenceNumber + 1;
@@ -1156,39 +1192,6 @@ std::shared_ptr<TransactionLogStore> TransactionLogStore::load(
 
 					sequenceNumber = std::stoul(sequenceNumberStr);
 
-					// check if the file is too old
-					if (retentionMs.count() > 0) {
-						auto mtime = std::filesystem::last_write_time(filePath);
-						auto mtime_sys = convertFileTimeToSystemTime(mtime);
-						auto now = std::chrono::system_clock::now();
-						auto fileAgeMs = std::chrono::duration_cast<std::chrono::milliseconds>(now - mtime_sys);
-						auto delta = fileAgeMs - retentionMs;
-
-						if (delta.count() > 0) {
-							// file is too old, remove it
-							DEBUG_LOG("%p TransactionLogStore::load File \"%s\" age=%lldms, expired %lldms ago, purging\n",
-								store.get(), filePath.filename().string().c_str(), fileAgeMs.count(), delta.count());
-							try {
-								DEBUG_LOG("%p TransactionLogStore::load Removing expired file: %s\n", store.get(), filePath.string().c_str());
-								if (std::filesystem::remove(filePath)) {
-									std::error_code markerError;
-									auto markerPath = transactionLogAppendBoundaryMarkerPath(filePath);
-									std::filesystem::remove(markerPath, markerError);
-									std::filesystem::remove(markerPath.parent_path(), markerError);
-									std::filesystem::remove(
-										markerPath.parent_path().parent_path(), markerError);
-								}
-							} catch (const std::filesystem::filesystem_error& e) {
-								DEBUG_LOG("%p TransactionLogStore::load Failed to remove expired file %s: %s\n",
-									store.get(), filePath.string().c_str(), e.what());
-							}
-							continue;
-						} else {
-							DEBUG_LOG("%p TransactionLogStore::load File \"%s\" age=%lldms, not expired, %lldms left\n",
-								store.get(), filePath.filename().string().c_str(), fileAgeMs.count(), delta.count() * -1);
-						}
-					}
-
 					store->registerLogFile(filePath, sequenceNumber);
 				}
 			} catch (const TransactionLogAppendBoundaryException&) {
@@ -1222,14 +1225,40 @@ std::shared_ptr<TransactionLogStore> TransactionLogStore::load(
 		auto currentIt = store->sequenceFiles.find(storeCurrentSeq);
 		if (currentIt != store->sequenceFiles.end()) {
 			auto& currentFile = currentIt->second;
-			if (!currentFile->isOpen()) {
-				currentFile->open(store->latestTimestamp);
+			currentFile->appendBoundaryMarkerEnabled = true;
+			bool activated = true;
+			try {
+				if (!currentFile->isOpen()) {
+					currentFile->open(store->latestTimestamp);
+				}
+			} catch (const TransactionLogAppendBoundaryException&) {
+				// The marker is authoritative; a segment we cannot reconcile with it
+				// must fail the load rather than expose orphaned bytes.
+				throw;
+			} catch (const std::exception& e) {
+				// A newest segment we cannot open is not fatal: leave it registered
+				// (readers still see whatever a later ensureExtent() can resolve) and
+				// start appending at a fresh sequence instead of failing the open.
+				DEBUG_LOG("%p TransactionLogStore::load Failed to open current log file %s: %s\n",
+					store.get(), currentFile->path.string().c_str(), e.what());
+				currentFile->appendBoundaryMarkerEnabled = false;
+				activated = false;
+			}
+			if (activated) {
+				uint32_t protectedPosition = flushedPosition.logSequenceNumber == storeCurrentSeq
+					? flushedPosition.positionInLogFile
+					: 0;
+				currentFile->recoverTail(protectedPosition);
+				store->nextLogPosition = { currentFile->size, storeCurrentSeq };
+			} else {
+				uint32_t nextWritableSequence = storeCurrentSeq + 1;
+				store->currentSequenceNumber.store(nextWritableSequence, std::memory_order_relaxed);
+				store->nextLogPosition = { 0, nextWritableSequence };
+				if (store->nextSequenceNumber <= nextWritableSequence) {
+					store->nextSequenceNumber = nextWritableSequence + 1;
+				}
+				storeCurrentSeq = nextWritableSequence;
 			}
-			uint32_t protectedPosition = flushedPosition.logSequenceNumber == storeCurrentSeq
-				? flushedPosition.positionInLogFile
-				: 0;
-			currentFile->recoverTail(protectedPosition);
-			store->nextLogPosition = { currentFile->size, storeCurrentSeq };
 		}
 	}
 
@@ -1284,6 +1313,15 @@ std::shared_ptr<TransactionLogStore> TransactionLogStore::load(
 		}
 	}
 	*store->lastCommittedPosition = recoveredPosition < flushedPosition ? flushedPosition : recoveredPosition;
+	if (retentionMs.count() > 0) {
+		try {
+			store->purge(nullptr, false, 0, false);
+		} catch (const std::exception& e) {
+			DEBUG_LOG("%p TransactionLogStore::load Failed to purge expired files: %s\n", store.get(), e.what());
+		} catch (...) {
+			DEBUG_LOG("%p TransactionLogStore::load Failed to purge expired files\n", store.get());
+		}
+	}
 
 	return store;
 }

src/binding/transaction_log/transaction_log_store.h

@@ -546,10 +546,10 @@ struct TransactionLogStore final {
 	 * Resolves `file->size` — the written extent — for a file that was
 	 * registered but never opened, by borrowing its handle briefly.
 	 *
-	 * registerLogFile() opens the current file eagerly but leaves older ones
-	 * lazy, and a lazy file reads `size == 0`, which is indistinguishable from
-	 * "empty" to every consumer of the field. Two of them act on that: the
-	 * backup snapshot would omit the segment, and doPurge()'s flushed-position
+	 * Discovery leaves every registered file closed (load() opens only the
+	 * surviving current one), and a closed file reads `size == 0`, which is
+	 * indistinguishable from "empty" to every consumer of the field. Two of them
+	 * act on that: the backup snapshot would omit the segment, and doPurge()'s flushed-position
 	 * guard would delete a segment whose tail never reached RocksDB. Anything
 	 * that makes a decision from `size` must call this first.
 	 *

test/transaction-log-stats.test.ts

@@ -114,6 +114,24 @@ describe('Transaction Log Stats', () => {
 				}
 			));
 
+		it('should not report the flushed current file as purgeable', () =>
+			dbRunner(
+				{ dbOptions: [{ path: generateDBPath(), transactionLogRetention: 500 }] },
+				async ({ db }) => {
+					const log = db.useLog('retain-current');
+					await db.transaction(async (txn) => {
+						log.addEntry(Buffer.alloc(100, 'a'), txn.id);
+						db.putSync('key', 'value', { transaction: txn });
+					});
+					db.flushSync();
+					await delay(1200);
+
+					const stats = log.getStats();
+					expect(stats.purge.purgeableFiles).toBe(0);
+					expect(stats.purge.retainedUnflushedFiles).toBe(0);
+				}
+			));
+
 		it('should increment purgeRuns when logs are purged', () =>
 			dbRunner(async ({ db }) => {
 				const log = db.useLog('purge');