Skip a replicated publish the receiving node has already logged, so a restart or reconnect doesn't deliver it twice

docs/release-notes/v5-lincoln/5.4/replicated-messages-arrive-once-after-a-restart.md

@@ -0,0 +1,7 @@
+---
+section: 'Replication'
+---
+
+### Replicated Messages Arrive Once After a Restart or Reconnect
+
+After a node restarted, restarted its HTTP workers or reconnected replication, its subscribers could receive a burst of messages published on another node a second time. A resuming node asks its peer to re-send the part of the log just before what it had applied, so that a transaction committed late is not missed; record writes in that stretch were already dropped as duplicates, but published messages were applied and delivered again. On RocksDB, a replicated message the node's transaction log still holds is now skipped, and a message it never received is still delivered once; only a message at or below its topic's record version is looked up (or, on a table that evicts or expires records, one whose record is gone), so in-order messages cost nothing extra. Nothing needs to change in applications or configuration. A message is still applied again when it is re-sent after audit retention purged its entry, on a node configured without transaction-log abort markers, or when its topic's record was moved back by an earlier version before upgrading.

resources/DESIGN.md

@@ -22,6 +22,7 @@ Declarations compile to `fullTextIndexes`, never `attributes` (no stored value).
 - `receivedIdentityRefs` holds received identities, never log keys: unaudited, the only re-delivery proof ([receipt rules](receipt-rules.md)). **Never "fix" a log key to the version** (`auditRefKinds.test.js`).
 - LMDB merges keep the older version (`noAuditOutOfOrderPatch.test.js`). Duplicates use committed identity; only minted keys mark a retry's own staging (`readsOwnStaging`, `replicatedRetryDuplicate.test.js`).
 - RocksDB's watermark stalls at uncommitted entries, so ref walks pass `followingCommittedRef`. Gap: history walks follow one chain, plus late messages (#3977).
+- A replicated publish is skipped only on its logged entry (`isMessageLogged`), at or below the record's version, with abort markers on; in history older than the process only a record vouches for it (`replicatedPublishRedelivery.test.js`).
 - **A record's version never moves back**: a message older than it keeps the version and refs its entry, which history walks reach by `isLateEntry` (`latePublishVersionRegression.test.js`).
 - `FULL_WRITE_HEAD`: the head replaced the record and is unchanged since; any other write clears it (`fullWriteHead.test.js`).
 - Named-log misses never fall back to `local`; `local` reads filter by `nodeId` (`auditLog.test.js`).

resources/Table.ts

@@ -223,7 +223,7 @@ import {
 	GET_TRACKED_WRITE_GENERATION,
 } from './tracked.ts';
 import { transaction, contextStorage } from './transaction.ts';
-import { MAXIMUM_KEY, writeKey, compareKeys } from 'ordered-binary';
+import { MAXIMUM_KEY, writeKey, compareKeys, toBufferKey } from 'ordered-binary';
 import {
 	getWorkerIndex,
 	applicationWorkerIndex,
@@ -251,6 +251,7 @@ import {
 	auditHistoryAbortMarked,
 	type AuditRecord,
 } from './auditStore.ts';
+import { abortMarkersEnabled } from './transactionLogAbortMarkers.ts';
 import {
 	acquireFullTextClearFence,
 	acquireFullTextRetirementFence,
@@ -389,6 +390,10 @@ type MaybePromise<T> = T | Promise<T>;
 const NULL_WITH_TIMESTAMP = new Uint8Array(9);
 NULL_WITH_TIMESTAMP[8] = 0xc0; // null
 const sourceWriteTypes = new Set(['put', 'patch', 'delete', 'publish', 'message', 'invalidate', 'relocate']);
+const loggedMessagesByTransaction = new WeakMap<object, Map<string, Set<string> | null>>();
+// the storage-key encoding, so ids the store treats as one key match
+const messageIdentity = (tableId: number, id: Id, version: number) =>
+	`${tableId}:${version}:${toBufferKey(id).toString('latin1')}`;
 const isSourceWriteType = (type: string) => sourceWriteTypes.has(type);
 // declared types `validate` holds to one value on validated writes: `ID` admits an array of strings
 const SCALAR_TYPES = new Set(['String', 'Int', 'Long', 'Float', 'Boolean', 'Date', 'BigInt']);
@@ -11351,10 +11356,30 @@ export function makeTable(options): TableResourceClass {
 					(this.constructor as any).source?.publish && !(context as any)?.source
 						? (this.constructor as any).source.publish.bind((this.constructor as any).source, id, message, context)
 						: undefined,
-				commit: (txnTime, existingEntry, _retry, transaction: any) => {
+				commit: (txnTime, existingEntry, retry, transaction: any, committedBy?: DatabaseTransaction) => {
+					write.skipped = false;
 					write.stagedEntry = undefined;
 					write.completeResidencyRecord = undefined;
 					write.superseded = false;
+					const precedes =
+						existingEntry?.version !== undefined ? precedesExistingVersion(txnTime, existingEntry, options?.nodeId) : 1;
+					// The record's version never moves back, so only a message at or below it, or one to a record the table can
+					// drop, can have been logged here, and the logged entry is the proof. A retry recommits the batch its first
+					// attempt staged; an earlier process's file can hold an unmarked aborted batch, so there only a record vouches.
+					if (
+						committedBy?.sourceApply &&
+						!committedBy.isReplay &&
+						isRocksDB &&
+						abortMarkersEnabled() &&
+						!(retry && write.appendedAuditEntry === true) &&
+						(existingEntry === undefined
+							? (evictsRecords() || !audit) && auditHistoryAbortMarked(auditStore, options?.nodeId)
+							: precedes <= 0) &&
+						isMessageLogged(committedBy, (context as any)?.timestamp, id, options?.nodeId, options?.version ?? txnTime)
+					) {
+						write.skipped = true;
+						return;
+					}
 					if (existingEntry === undefined && trackDeletes && !audit) {
 						scheduleCleanup();
 					}
@@ -11365,9 +11390,7 @@ export function makeTable(options): TableResourceClass {
 					const base = priorStagedOp?.stagedEntry ?? existingEntry;
 					const value = base?.value ?? null;
 					const metadataFlags = (base?.metadataFlags ?? 0) & (LOCAL_ONLY | INVALIDATED | REPLICA);
-					const late =
-						existingEntry?.version !== undefined &&
-						precedesExistingVersion(txnTime, existingEntry, options?.nodeId) < 0;
+					const late = precedes < 0;
 					const ownRefs = late
 						? isRocksDB
 							? withAuditRef(existingEntry.additionalAuditRefs, appendedLogKey, options?.nodeId)
@@ -14393,6 +14416,38 @@ export function makeTable(options): TableResourceClass {
 		return Boolean(entry?.metadataFlags & INVALIDATED) && isAuditEntryWrite(entry, { txnLogKey, nodeId });
 	}
 
+	/**
+	 * An origin transaction commits here all at once, so one read answers for all its messages for the applying
+	 * transaction; no write that staged its entry asks again on a retry.
+	 */
+	function isMessageLogged(
+		applying: DatabaseTransaction,
+		txnLogKey: number | undefined,
+		id: Id,
+		nodeId: number | undefined,
+		version: number
+	): boolean {
+		if (txnLogKey === undefined || !nodeId) return false;
+		// an unnamed origin's entries sit in `local` under minted keys, never at its received key
+		if (!auditStore.logById(nodeId)) return false;
+		let byOriginKey = loggedMessagesByTransaction.get(applying);
+		if (!byOriginKey) loggedMessagesByTransaction.set(applying, (byOriginKey = new Map()));
+		const originKey = `${nodeId}:${txnLogKey}`;
+		let logged = byOriginKey.get(originKey);
+		if (logged === undefined) byOriginKey.set(originKey, (logged = loggedMessagesAt(txnLogKey, nodeId)));
+		return logged !== null && logged.has(messageIdentity(tableId, id, version));
+	}
+
+	function loggedMessagesAt(txnLogKey: number, nodeId: number): Set<string> | null {
+		let logged: Set<string> | null = null;
+		for (const entry of auditStore.getRange({ start: txnLogKey, exactStart: true, log: nodeId })) {
+			if (entry.txnLogKey !== txnLogKey) break;
+			if (entry.type === 'message')
+				(logged ??= new Set()).add(messageIdentity(entry.tableId, entry.recordId, entry.version));
+		}
+		return logged;
+	}
+
 	// Not stagedEntry: a relocate restamps the record without staging it, and a publish's restamp counts too.
 	function followsAppliedWrite(write: any): boolean {
 		for (let prior = write.priorWrite; prior; prior = prior.priorWrite) {

unitTests/resources/relayedHeadTieDuplicate.test.js

@@ -237,16 +237,22 @@ describe('Relayed head-tie duplicates (harper#3467)', function () {
 			],
 		});
 		await apply(transaction());
-		const puts = async () => (await T.getHistoryOfRecord('r')).filter((entry) => entry.type === 'put').length;
-		const putsBefore = await puts();
+		const logged = async (type) => (await T.getHistoryOfRecord('r')).filter((entry) => entry.type === type).length;
+		const putsBefore = await logged('put');
+		const messagesBefore = await logged('message');
 		const spy = spyRecordLogReads(T, version, origin);
 		try {
 			await apply(transaction());
 		} finally {
 			spy.restore();
 		}
-		assert.deepEqual(spy.reads, [], 'the publish runs after the put, so the put reads no committed base');
-		assert.equal(await puts(), putsBefore, 'the re-delivered put must not be logged again');
+		assert.deepEqual(
+			spy.reads,
+			[{ getRange: version, log: origin }],
+			'the publish runs after the put, so the put reads no committed base; only the publish looks up its entry'
+		);
+		assert.equal(await logged('put'), putsBefore, 'the re-delivered put must not be logged again');
+		assert.equal(await logged('message'), messagesBefore, 'the re-delivered publish must not be logged again');
 		assert.equal((await T.get('r')).name, 'first');
 	});