Stamp created and updated times at the version a refused, re-logged retry commits under

resources/DESIGN.md

@@ -162,16 +162,16 @@ Pre-commit blob files land before a commit that can still skip the write.
 
 ## Over-time transactions are aborted, not force-committed
 
-`startMonitoringTxns` acts past `STORAGE_MAXTRANSACTIONOPENTIME`. Code committing a native handle itself opens it raw, never via a wrapper's `getReadTxn`: the wrapper would stay tracked, and its retry would resolve `RETRY_NOW` rather than reject with `ERR_BUSY`.
+`startMonitoringTxns` acts past `STORAGE_MAXTRANSACTIONOPENTIME`. Code committing a native handle itself opens it raw, never via a wrapper's `getReadTxn`: the wrapper would stay tracked, and its retry would resolve `RETRY_NOW`, not reject with `ERR_BUSY`.
 
 - **A write-bearing transaction is aborted and poisoned** (422 `transactionOpenTooLongError`): force-committing a partial write set breaks atomicity. `hasPendingWrites()` walks `next`.
-- Read-only, `sourceApply` and `isReplay` transactions are force-committed. Sources and replay cannot resume, so an abort drops a write the cursor passed (harper-pro#348).
+- Read-only, `sourceApply` and `isReplay` transactions are force-committed: sources and replay cannot resume, so an abort drops a write the cursor passed (harper-pro#348).
 - The owner's commit chains on `monitorCommit`. The pre-commit phase is spared `COMMIT_PHASE_GRACE` ticks (`sourceApply`/`isReplay` without limit): its writes are sealed, and poisoning there once reported success after unlinking blobs.
 - After the pre-commit await, the commit rechecks liveness or resolves as a phantom. **The monitor never aborts a submitted native commit.**
-- `timeoutBudget` is a sticky floor that chained links inherit.
+- `timeoutBudget` is a sticky floor chained links inherit.
 - Conflict retries get one deadline per logical commit (503 `TransactionCommitConflictTimeoutError`); the per-attempt clock (`trackOutstandingCommit`) drives `checkOverloaded()` shedding.
-- **Abort markers let a retry abort a logged handle and dedup trust its log** (harper#3236; `mustRetryLoggedCommit`, `mayReadOwnOrphan`). After a restart dedup waits ~3d for older log files to purge; meanwhile a twin delivery may double-count.
-- Each hop of a sealed commit cascade gets a full idle window, never the previous hop's remainder (`renewChainForNativeCommit`).
+- **Abort markers let a retry abort a logged handle and dedup trust its log** (harper#3236; `mustRetryLoggedCommit`, `mayReadOwnOrphan`); until old log files purge (~3d after a restart) a twin may double-count. Re-logging restamps record times (`stampRecordTimes`).
+- Each hop of a sealed commit cascade gets a full idle window, not the last hop's remainder (`renewChainForNativeCommit`).
 - Long-lived handles are reported from main only: rocksdb-js's registry is process-global (`runLongLivedTransactionSweep`, past `storage.longTransactionReportThreshold`); workers attribute their own by the same id (`reportIfLongLived`). Stuck-commit logs list holders from every database regardless of age: the verification table is one process-wide array (`longLivedTransactions.test.js`).
 
 ## A client disconnect cancels the request's writes

resources/Table.ts

@@ -1795,6 +1795,18 @@ export function makeTable(options): TableResourceClass {
 		if (missingCreatedTime) completed[createdTimeProperty.name] = timestampValue(createdTimeProperty, version);
 		return Object.freeze(completed);
 	}
+	function stampRecordTimes(record: any, base: Partial<Entry> | undefined, version: number, fullUpdate: boolean) {
+		if (updatedTimeProperty) record[updatedTimeProperty.name] = timestampValue(updatedTimeProperty, version);
+		if (!createdTimeProperty) return;
+		// a cache stub without a created time was never stored; a residency stub's record lives on its residents
+		if (
+			base?.value &&
+			(!(base.metadataFlags & INVALIDATED) || base.residencyId || base.value[createdTimeProperty.name] != null)
+		) {
+			if (fullUpdate || record[createdTimeProperty.name])
+				record[createdTimeProperty.name] = base.value[createdTimeProperty.name];
+		} else record[createdTimeProperty.name] = timestampValue(createdTimeProperty, version);
+	}
 	const tableGeneration = options.storageGeneration ?? (primaryKeyAttribute as any)?.generation;
 	const createdTime: number | undefined = options.createdTime ?? (primaryKeyAttribute as any)?.createdTime;
 	let deleteCallbackHandle: { remove: () => void };
@@ -5853,29 +5865,9 @@ export function makeTable(options): TableResourceClass {
 								this.validate(recordUpdate, !fullUpdate);
 								if (ownMerged) applyOwnChanges(this, recordUpdate, ownKeys);
 							}
-							if (updatedTimeProperty) {
-								recordUpdate[updatedTimeProperty.name] = timestampValue(updatedTimeProperty, txnTime);
-							}
-							if (createdTimeProperty) {
-								// the reloaded commit base, not the pre-read one: a full PUT racing a create
-								// would otherwise stamp a fresh created time over the real one
-								const base = write.entry;
-								// a cache stub without a created time was never stored; a residency stub's record lives on its residents
-								if (
-									base?.value &&
-									(!(base.metadataFlags & INVALIDATED) ||
-										base.residencyId ||
-										base.value[createdTimeProperty.name] != null)
-								) {
-									if (fullUpdate || recordUpdate[createdTimeProperty.name]) {
-										// make sure to retain original created time
-										recordUpdate[createdTimeProperty.name] = base.value[createdTimeProperty.name];
-									}
-								} else {
-									// new entry, set created time
-									recordUpdate[createdTimeProperty.name] = timestampValue(createdTimeProperty, txnTime);
-								}
-							}
+							// the reloaded commit base, not the pre-read one: a full PUT racing a create
+							// would otherwise stamp a fresh created time over the real one
+							stampRecordTimes(recordUpdate, write.entry, txnTime, fullUpdate);
 							if (primaryKey && recordUpdate[primaryKey] !== id && (fullUpdate || primaryKey in recordUpdate)) {
 								// ensure that the primary key is correct, if there is supposed to be one
 								recordUpdate[primaryKey] = id;
@@ -5933,6 +5925,20 @@ export function makeTable(options): TableResourceClass {
 					const priorStaged = priorStagedOp?.stagedEntry;
 					let existingRecord =
 						priorStagedOp?.completeResidencyRecord ?? (priorStaged ? priorStaged.value : existingEntry?.value);
+					// A batch re-logged after a refusal can commit under a new version (a fresh key) and onto a
+					// base the retry reloaded, neither of which validate stamped against.
+					if (
+						retry &&
+						transaction?.relogOnRetry === true &&
+						recordUpdate &&
+						!(context as any)?.source &&
+						(updatedTimeProperty || createdTimeProperty)
+					) {
+						const frozen = Object.isFrozen(recordUpdate);
+						recordUpdate = { ...recordUpdate };
+						stampRecordTimes(recordUpdate, priorStaged ?? existingEntry, txnTime, fullUpdate);
+						if (frozen) Object.freeze(recordUpdate);
+					}
 					if (databaseName === 'system' && tableName === RESIDENCY_MAP_TABLE) {
 						try {
 							if (!fullUpdate) throw new ClientError('A shard map must be written as a complete snapshot', 400);

resources/originClosedFloor.ts

@@ -413,9 +413,11 @@ export function reservedLocalKey(handle: RocksTransaction | undefined): number |
 }
 
 /**
- * A retry whose attempt wrote no log batch logs afresh, under a fresh key, so it stays in order with the
+ * A retry whose attempt wrote no log batch logs afresh, under a fresh key, so its version stays above the
  * writes that beat it; the published bound stays at the original, smaller key, which still bounds it.
- * Returns the new key, or undefined when the handle has no minted reservation or its timestamp is frozen.
+ * Returns the new key, or undefined when the handle has no reservation or reserved around a caller's value:
+ * that write's version is the caller's value whatever its key, so a fresh key would order nothing, and the
+ * key it reserved is still unused and above the floor.
  */
 export function rekeyLocalRetry(handle: RocksTransaction): number | undefined {
 	const reservation: Reservation | undefined = (handle as any)[RESERVATION];

unitTests/resources/abortMarkersAdoption.test.js

@@ -112,13 +112,12 @@ describe('Transaction-log abort markers (storage.rocks.transactionLogAbortMarker
 		});
 	}
 
-	async function relogAfterRefusal(targetTable) {
-		const recordId = id('relog');
-		await seed(targetTable, recordId);
+	// `holder` stages a write intent on `recordId`, which refuses `mine`'s first commit before its log write.
+	async function refuseOnHeldIntent(targetTable, recordId, { holderCommits, mineContext = {}, patch = false }) {
 		const targetDb = targetTable.primaryStore.store.db;
 		const originalCommit = Transaction.prototype.commit;
 		const attempts = [];
-		let holderId;
+		let holderId, holderVersion;
 		let submitted = false;
 		Transaction.prototype.commit = async function (...args) {
 			if (this.store?.db !== targetDb || this.id === holderId) return originalCommit.apply(this, args);
@@ -137,23 +136,36 @@ describe('Transaction-log abort markers (storage.rocks.transactionLogAbortMarker
 			const write = holderContext.transaction.writes.find((candidate) => candidate.key === recordId);
 			await holderContext.transaction.save(write);
 			holderId = holderContext.transaction.transaction.id;
+			holderVersion = holderContext.transaction.timestamp;
 			await holderGate;
-			throw new Error('holder gives up');
+			if (!holderCommits) throw new Error('holder gives up');
 		});
 		holder.catch(() => {});
 		await waitFor(() => holderId !== undefined);
-		const mine = writeAfterRead(targetTable, recordId, 'mine');
+		const mine = transaction(mineContext, async () => {
+			await targetTable.get(recordId, mineContext);
+			if (patch) await targetTable.patch(recordId, { writer: 'mine' }, mineContext);
+			else await targetTable.put({ id: recordId, writer: 'mine' }, mineContext);
+		});
 		await waitFor(() => submitted);
 		await delay(20); // the native intent check runs on the commit thread after submission
 		releaseHolder();
 		await mine;
-		await assert.rejects(holder, /holder gives up/);
+		if (holderCommits) await holder;
+		else await assert.rejects(holder, /holder gives up/);
 		assert.equal(attempts.length, 2);
 		assert.deepEqual(
 			{ kind: attempts[0].kind, logWritten: attempts[0].logWritten },
 			{ kind: ConflictKind.intentBeforeLog, logWritten: false }
 		);
 		assert.equal(attempts[1].logWritten, true);
+		return { attempts, holderVersion };
+	}
+
+	async function relogAfterRefusal(targetTable) {
+		const recordId = id('relog');
+		await seed(targetTable, recordId);
+		const { attempts } = await refuseOnHeldIntent(targetTable, recordId, { holderCommits: false });
 		assert.ok(attempts[1].key > attempts[0].key, 'the re-logged retry takes a fresh key');
 		const entries = auditEntries(targetTable, recordId);
 		assert.equal(entries.filter((entry) => entry.txnLogKey === attempts[0].key).length, 0);
@@ -304,6 +316,71 @@ describe('Transaction-log abort markers (storage.rocks.transactionLogAbortMarker
 		await relogAfterRefusal(Audited);
 	});
 
+	describe('a retry refused before its log write', function () {
+		let Stamped;
+		before(function () {
+			Stamped = table({
+				table: 'AbortMarkersStamped',
+				database: 'abortMarkersStamped',
+				attributes: [
+					{ name: 'id', isPrimaryKey: true },
+					{ name: 'writer' },
+					{ name: 'created', assignCreatedTime: true },
+					{ name: 'updated', assignUpdatedTime: true },
+				],
+				audit: true,
+			});
+		});
+		beforeEach(function () {
+			const priorIntents = setWriteIntentsForTests(true);
+			restores.push(() => setWriteIntentsForTests(priorIntents ?? false));
+		});
+
+		it('stamps the created and updated times of a create at the version it commits under', async function () {
+			const recordId = id('stamped-create');
+			const { attempts } = await refuseOnHeldIntent(Stamped, recordId, { holderCommits: false });
+			assert.ok(attempts[1].key > attempts[0].key, 'the re-logged retry takes a fresh key');
+			const entry = Stamped.primaryStore.getEntry(recordId);
+			assert.equal(entry.version, attempts[1].key);
+			assert.deepEqual(
+				{ writer: entry.value.writer, created: entry.value.created, updated: entry.value.updated },
+				{ writer: 'mine', created: entry.version, updated: entry.version }
+			);
+		});
+
+		for (const patch of [false, true]) {
+			it(`keeps the created time of the record a concurrent creator committed first (${patch ? 'patch' : 'put'})`, async function () {
+				const recordId = id('stamped-raced');
+				const { attempts, holderVersion } = await refuseOnHeldIntent(Stamped, recordId, { holderCommits: true, patch });
+				const entry = Stamped.primaryStore.getEntry(recordId);
+				assert.equal(entry.version, attempts[1].key);
+				assert.ok(entry.version > holderVersion, 'the retry commits after the holder');
+				assert.deepEqual(
+					{ writer: entry.value.writer, created: entry.value.created, updated: entry.value.updated },
+					{ writer: 'mine', created: holderVersion, updated: entry.version }
+				);
+			});
+		}
+
+		it('keeps the created time of a concurrent creator under an explicit timestamp', async function () {
+			const recordId = id('stamped-explicit');
+			const timestamp = Date.now() + 60000;
+			const { attempts, holderVersion } = await refuseOnHeldIntent(Stamped, recordId, {
+				holderCommits: true,
+				mineContext: { timestamp },
+			});
+			assert.equal(attempts[1].key, attempts[0].key);
+			const entry = Stamped.primaryStore.getEntry(recordId);
+			assert.equal(entry.version, timestamp);
+			assert.deepEqual(
+				{ writer: entry.value.writer, created: entry.value.created, updated: entry.value.updated },
+				{ writer: 'mine', created: holderVersion, updated: timestamp }
+			);
+			const entries = auditEntries(Stamped, recordId).filter((entry) => entry.txnLogKey === attempts[1].key);
+			assert.equal(entries.length, 1);
+		});
+	});
+
 	it('turns write intents on from storage.rocks.transactionLogWriteIntents', async function () {
 		const env = require('#src/utility/environment/environmentManager');
 		const configured = env.get('storage_rocks_transactionLogWriteIntents');