Expose record version and txnLogKey as separate audit clocks

dataLayer/harperBridge/ResourceBridge.ts

@@ -579,7 +579,9 @@ export class ResourceBridge extends BridgeMethods {
 						let operation = normalizeHistoryOperation(auditRecord.operation, auditRecord.type);
 						return {
 							operation,
-							timestamp: auditRecord.version,
+							// the transaction's position in the log, which is what this field has always
+							// reported on RocksDB and what groups an atomic transaction (harper#2412)
+							timestamp: auditRecord.localTime,
 							user_name: auditRecord.user,
 							ids: [id],
 							records: [auditRecord.value],
@@ -775,7 +777,7 @@ async function* groupRecordsInHistory(table, start?, end?, limit?) {
 	let count = 0;
 	for await (const entry of table.getHistory(start, end)) {
 		let operation = normalizeHistoryOperation(entry.operation, entry.type);
-		const { id, version: timestamp, value } = entry;
+		const { id, localTime: timestamp, value } = entry;
 		if (enqueued?.timestamp === timestamp) {
 			enqueued.ids.push(id);
 			enqueued.records.push(value);

resources/DESIGN.md

@@ -101,6 +101,50 @@ One giant `makeTable()` factory that returns a `TableResource extends Resource`
 | How does post-ordering resolve vector distances safely?               | Each comparator owns its `Sort`, passes it directly to the custom-index resolver, and caches distances by that immutable per-query sort object.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
 | How is application row filtering applied?                             | Authorization admission happens in the resource operation before query work. The legacy `allow*` hook, when armed by the protocol, is evaluated once with its historical receiver semantics; overriding it never changes its scope. An operation override may add indexed conditions and/or attach the JavaScript-only synchronous `target.rowFilter(record, context)`. `Table.search` composes it with query filters and rechecks the final materialized cache/source record. `SubscriptionRequest.rowFilter` covers full-row events; `eventFilter(event, context)` explicitly handles tombstones/messages/raw events. Prefer indexed conditions because an opaque predicate may inspect every admitted candidate and `limit` applies after filtering.                                                                                                                                                                                                                                                                                                                                                           |
 
+**An audit record carries two clocks; never substitute one for the other (harper#2412 stage 0b).**
+`AuditRecord.version` is the originating write's record version — LWW ordering in
+`precedesExistingVersion`, `@updatedTime`, ETag/`Last-Modified`. Historical audit-only entries may
+carry the surviving version instead. `AuditRecord.txnLogKey` is that
+entry's key in the per-origin transaction log: write identity, the record→log lookup
+(`auditStore.get(logKey, tableId, id, nodeId)`), and every resume cursor. On LMDB these have always
+been distinct fields; on RocksDB the read surface used to overwrite `version` with the log key, and
+`#2409`'s `recordVersion` alias is now absorbed back into `version`.
+
+They hold the same value for every write whose record version is its own commit timestamp, which is
+every ordinary local write, so a bug that confuses them stays invisible until a **source fill**
+(`getFromSource`, core #2065): the record is stored at the source-reported version while its log
+entry is keyed at the fill's commit. A RocksDB record stores its version in the existing word and,
+when the clocks diverge, keeps its audit head in `additionalAuditRefs`; stage 2 can replace that
+compatibility pointer with a dedicated log-key word.
+
+Consequences worth knowing:
+
+- **Identity is `(nodeId, log key)`, never a version.** `isAuditEntryWrite` (`auditStore.ts`) is the
+  single predicate; `removeAuditEntry`'s tombstone removal and `blob.ts`'s orphan sweep both gate on
+  it, and both retain rather than delete when identity is unknown. A version compare there would let
+  one write authorize destroying another's tombstone or blob.
+- **A RocksDB applied write carries its own record version.** `TransactionWrite.recordVersion`, set from
+  `options.version` by every `_write*` builder and read in `save()` only when the transaction is
+  `sourceApply` or `isReplay`, is how a replication receiver stores the origin's version while the
+  transaction commits under the origin's log key — so a peer's copy of an origin's log stays in the
+  origin's clock. New audit entries persist that write version even when an out-of-order merge leaves
+  a newer record version in the primary store. `getAppliedWriteVersion` also bounds historical
+  overloaded values by `txnLogKey`. One frame can carry writes at different record versions, so this
+  cannot be a per-transaction value. Deprecated LMDB keeps its legacy transaction-version apply
+  behavior.
+- **`additionalAuditRefs[].version` is a log key, not a version.** Every consumer follows it straight
+  into `auditStore.get` (`Table.ts`'s `auditRefsToVisit`). The list carries folded out-of-order
+  branches and, when a RocksDB record version differs from its log key, the record's own audit head.
+  Under stage 2 the stored reference field can be renamed; until then, "fixing" it to the record
+  version silently unaddresses the entry it points at.
+- **Crash replay uses both.** `replayLogs` delimits transactions by `txnLogKey` (which is also what
+  `CorruptFrameStop.truncatedVersions` records) and replays each write at its stored `version`.
+  Stamping a replayed record at its log key would move its version forward and make a later
+  legitimate write look stale.
+- **Compatibility surfaces still report `localTime` as the transaction timestamp.** Subscription,
+  history, and pro-to-core replication event shapes predate this internal name and remain unchanged;
+  no on-disk or on-wire identifier changes in this stage.
+
 **QUERY admission uses the body projection.** `Resource.transactional` resolves an asynchronous HTTP QUERY body before resource resolution, recursively clones away client-supplied `checkPermission`, and copies the body's `select` onto the operation admission target before `allowRead`. After authorization, `Resource.query` transfers only the narrowed projection to the body target. This ensures a body-only relationship select is checked before `Table.search`; permission-control fields from QUERY data must never reach a search target.
 
 **Async false-mode read gates preserve the streaming contract.** `Table.search` returns an `ExtendedIterable` carrying the internal `SEARCH_AUTHORIZATION` promise. Static `Resource.search` and `query` await that verdict before returning a response; on success the wrapper initializes the real search before the transaction settles so its normal read snapshot stays reserved until iteration completes. The marker follows supported iterable transforms and retains `selectApplied`/`getColumns`, so async or mapped delegation cannot turn a denial into a truncated successful response.

resources/DatabaseTransaction.ts

@@ -312,6 +312,9 @@ export type TransactionWrite = {
 	// overload accounting, the replay marker and a no-op write's removal all belong to the committer.
 	validate?: (txnTime: number, committedBy: DatabaseTransaction) => void;
 	fullUpdate?: boolean;
+	// The origin record version carried by an applied or replayed write. Bound it by the origin's
+	// transaction-log key so malformed or historical overloaded values cannot move ordering past the write.
+	recordVersion?: number;
 	saved?: boolean;
 	deferSave?: boolean;
 	skipReplicationConfirmation?: boolean;
@@ -371,6 +374,10 @@ export type TransactionWrite = {
 	innerCommit?: MaybePromise<CommitResolution>;
 };
 
+export function getAppliedWriteVersion(recordVersion: number | undefined, txnLogKey: number): number {
+	return recordVersion == null ? txnLogKey : Math.min(recordVersion, txnLogKey);
+}
+
 /**
  * The state a preceding write in this transaction left for `operation`'s key, or undefined if this
  * is the first write to it. Within a transaction the writes are ordered by program order, and
@@ -1026,13 +1033,19 @@ export class DatabaseTransaction implements Transaction {
 			(transaction as RocksTransactionWithRetry).isRetry = true;
 		}
 		if (!txnTime) txnTime = this.timestamp = transaction.getTimestamp();
+		// `txnTime` is this transaction's timestamp — the key its entries take in the per-origin log.
+		// A write applied from elsewhere carries the origin's record version too, and that is what the
+		// record is stored at; the two coincide for every locally-originated write. Gated on the apply
+		// flags so an ordinary write never reads the property (harper#2412).
+		const writeVersion =
+			this.sourceApply || this.isReplay ? getAppliedWriteVersion(operation.recordVersion, txnTime) : txnTime;
 		if (reloadEntry || operation.entry === undefined) {
 			operation.entry = operation.store.getEntry(operation.key, { transaction });
 		}
 		if (!operation.saved) {
 			operation.saved = true;
 			// immediately execute in this transaction
-			if ((operation.validate?.(txnTime, this) as any) === false) {
+			if ((operation.validate?.(writeVersion, this) as any) === false) {
 				operation.commit = () => {}; // noop if we try again
 				return;
 			}
@@ -1043,9 +1056,9 @@ export class DatabaseTransaction implements Transaction {
 		}
 		if (lockHandle || this.recordLocks) operation.trackRecordVersion = true;
 		if (operation.trackRecordVersion) operation.recordVersionApplied = false;
-		const completion = operation.commit(txnTime, operation.entry, this.retries > 0, transaction) as Promise<void>;
+		const completion = operation.commit(writeVersion, operation.entry, this.retries > 0, transaction) as Promise<void>;
 		if (operation.trackRecordVersion)
-			operation.appliedRecordVersion = operation.recordVersionApplied ? txnTime : undefined;
+			operation.appliedRecordVersion = operation.recordVersionApplied ? writeVersion : undefined;
 		if (typeof completion?.then === 'function') this.stageCompletion(completion);
 		// Sticky record that THIS write staged with its audit entry appended (log entries batch on the
 		// native transaction and are durably written by its commit attempt — even a failed one — so

resources/RecordEncoder.ts

@@ -908,9 +908,9 @@ export function recordUpdater(store, tableId, auditStore) {
 				metadataInNextEncoding |= HAS_RESIDENCY_ID;
 				extendedType |= HAS_CURRENT_RESIDENCY_ID;
 			} else residencyIdAtNextEncoding = 0;
-			const nodeId = options?.nodeId ?? (audit ? getThisNodeId(auditStore) : undefined);
-			if (nodeId >= 0) {
-				nodeIdAtNextEncoding = nodeId;
+			const recordNodeId = options?.recordNodeId ?? options?.nodeId ?? (audit ? getThisNodeId(auditStore) : undefined);
+			if (recordNodeId >= 0) {
+				nodeIdAtNextEncoding = recordNodeId;
 				metadataInNextEncoding |= HAS_NODE_ID;
 			} else nodeIdAtNextEncoding = -1;
 			const additionalAuditRefs = options?.additionalAuditRefs;
@@ -983,8 +983,22 @@ export function recordUpdater(store, tableId, auditStore) {
 				const nodeId = options?.nodeId ?? getThisNodeId(auditStore) ?? 0;
 				const viaNodeId = options?.viaNodeId ?? nodeId;
 				if (resolveRecord && existingEntry?.localTime) {
-					const replacingId = existingEntry?.localTime;
-					const replacingEntry = auditStore.get(replacingId, tableId, id);
+					let replacingId = existingEntry.localTime;
+					let replacingEntry;
+					if (isRocksDB && existingEntry.additionalAuditRefs) {
+						for (const ref of existingEntry.additionalAuditRefs) {
+							const candidate = auditStore.get(ref.version, tableId, id, ref.nodeId);
+							if (
+								candidate?.version === existingEntry.version &&
+								(candidate.nodeId ?? 0) === (existingEntry.nodeId ?? 0)
+							) {
+								replacingId = ref.version;
+								replacingEntry = candidate;
+								break;
+							}
+						}
+					}
+					replacingEntry ??= auditStore.get(replacingId, tableId, id, existingEntry.nodeId);
 					if (replacingEntry) {
 						const previousVersion = replacingEntry.previousVersion;
 						result = auditStore[isRocksDB ? 'putSync' : 'put'](
@@ -1013,7 +1027,7 @@ export function recordUpdater(store, tableId, auditStore) {
 				result = auditStore[isRocksDB ? 'putSync' : 'put'](
 					record === undefined ? NEW_TIMESTAMP_PLACEHOLDER : LAST_TIMESTAMP_PLACEHOLDER,
 					{
-						version: newVersion,
+						version: options?.recordVersion ?? newVersion,
 						tableId,
 						recordId: id,
 						previousVersion: isRocksDB ? existingEntry?.version : existingEntry?.localTime,

resources/RocksTransactionLogStore.ts

@@ -131,9 +131,8 @@ export class RocksTransactionLogStore extends EventEmitter {
 		}
 		if (this.listenerCount('aftercommit')) {
 			if (!(auditRecord instanceof Uint8Array)) {
-				const txnTimestamp = options.transaction.getTimestamp?.();
-				if (txnTimestamp != null) auditRecord.localTime = txnTimestamp;
-				auditRecord.recordVersion = auditRecord.version;
+				const txnLogKey = options.transaction.getTimestamp?.();
+				if (txnLogKey != null) auditRecord.txnLogKey = txnLogKey;
 			}
 			(options.transaction.logEntries ??= []).push(auditRecord);
 		}
@@ -184,7 +183,7 @@ export class RocksTransactionLogStore extends EventEmitter {
 				if (entry.recordId === recordId && entry.tableId === tableId) {
 					return entry;
 				}
-				if (entry.version !== key) return; // no longer in this transaction
+				if (entry.txnLogKey !== key) return; // no longer in this transaction
 			}
 		} else {
 			// Harper puts some metadata in the database, we will just put this in the root store instead
@@ -479,10 +478,7 @@ export class RocksTransactionLogStore extends EventEmitter {
 					position += 8;
 				}
 				const auditRecord = readAuditEntry(data, position, undefined);
-				// version stays the log key (replication resume relies on version === log key);
-				// the record's own version survives as recordVersion
-				auditRecord.version = timestamp;
-				auditRecord.localTime = timestamp;
+				auditRecord.txnLogKey = timestamp;
 				auditRecord.endTxn = endTxn;
 				auditRecord.previousResidencyId = previousResidencyId;
 				auditRecord.previousVersion = previousVersion;
@@ -494,7 +490,9 @@ export class RocksTransactionLogStore extends EventEmitter {
 					byteLength: data?.byteLength,
 				});
 				return {
+					// the log key is all this entry still yields; its record version is undecodable
 					version: timestamp,
+					txnLogKey: timestamp,
 					endTxn,
 					type: undefined,
 					tableId: undefined,