Merge pull request #559 from dao-xyz/fix/waitforreplicator-request-config

shared-log: make waitForReplicator request loop configurable

Author: marcus-pousette

packages/programs/data/shared-log/src/index.ts

@@ -366,6 +366,8 @@ export type SharedLogOptions<
 	syncronizer?: SynchronizerConstructor<R>;
 	timeUntilRoleMaturity?: number;
 	waitForReplicatorTimeout?: number;
+	waitForReplicatorRequestIntervalMs?: number;
+	waitForReplicatorRequestMaxAttempts?: number;
 	waitForPruneDelay?: number;
 	distributionDebounceTime?: number;
 	compatibility?: number;
@@ -376,6 +378,8 @@ export type SharedLogOptions<
 export const DEFAULT_MIN_REPLICAS = 2;
 export const WAIT_FOR_REPLICATOR_TIMEOUT = 9000;
 export const WAIT_FOR_ROLE_MATURITY = 5000;
+export const WAIT_FOR_REPLICATOR_REQUEST_INTERVAL = 1000;
+export const WAIT_FOR_REPLICATOR_REQUEST_MIN_ATTEMPTS = 3;
 // TODO(prune): Investigate if/when a non-zero prune delay is required for correctness
 // (e.g. responsibility/replication-info message reordering in multi-peer scenarios).
 // Prefer making pruning robust without timing-based heuristics.
@@ -564,6 +568,8 @@ export class SharedLog<
 
 	timeUntilRoleMaturity!: number;
 	waitForReplicatorTimeout!: number;
+	waitForReplicatorRequestIntervalMs!: number;
+	waitForReplicatorRequestMaxAttempts?: number;
 	waitForPruneDelay!: number;
 	distributionDebounceTime!: number;
 
@@ -1903,12 +1909,31 @@ export class SharedLog<
 			options?.timeUntilRoleMaturity ?? WAIT_FOR_ROLE_MATURITY;
 		this.waitForReplicatorTimeout =
 			options?.waitForReplicatorTimeout ?? WAIT_FOR_REPLICATOR_TIMEOUT;
+		this.waitForReplicatorRequestIntervalMs =
+			options?.waitForReplicatorRequestIntervalMs ??
+			WAIT_FOR_REPLICATOR_REQUEST_INTERVAL;
+		this.waitForReplicatorRequestMaxAttempts =
+			options?.waitForReplicatorRequestMaxAttempts;
 		this.waitForPruneDelay = options?.waitForPruneDelay ?? WAIT_FOR_PRUNE_DELAY;
 
 		if (this.waitForReplicatorTimeout < this.timeUntilRoleMaturity) {
 			this.waitForReplicatorTimeout = this.timeUntilRoleMaturity; // does not makes sense to expect a replicator to mature faster than it is reachable
 		}
 
+		if (this.waitForReplicatorRequestIntervalMs <= 0) {
+			throw new Error(
+				"waitForReplicatorRequestIntervalMs must be a positive number",
+			);
+		}
+		if (
+			this.waitForReplicatorRequestMaxAttempts != null &&
+			this.waitForReplicatorRequestMaxAttempts <= 0
+		) {
+			throw new Error(
+				"waitForReplicatorRequestMaxAttempts must be a positive number",
+			);
+		}
+
 		this._closeController = new AbortController();
 		this._isTrustedReplicator = options?.canReplicate;
 		this.keep = options?.keep;
@@ -3360,11 +3385,13 @@ export class SharedLog<
 		}, timeoutMs);
 
 		let requestAttempts = 0;
-		const requestIntervalMs = 1000;
-		const maxRequestAttempts = Math.max(
-			3,
-			Math.ceil(timeoutMs / requestIntervalMs),
-		);
+		const requestIntervalMs = this.waitForReplicatorRequestIntervalMs;
+		const maxRequestAttempts =
+			this.waitForReplicatorRequestMaxAttempts ??
+			Math.max(
+				WAIT_FOR_REPLICATOR_REQUEST_MIN_ATTEMPTS,
+				Math.ceil(timeoutMs / requestIntervalMs),
+			);
 
 		const requestReplicationInfo = () => {
 			if (settled || this.closed) {

packages/programs/data/shared-log/test/utils/stores/event-store.ts

@@ -62,6 +62,8 @@ export type Args<
 	timeUntilRoleMaturity?: number;
 	waitForPruneDelay?: number;
 	waitForReplicatorTimeout?: number;
+	waitForReplicatorRequestIntervalMs?: number;
+	waitForReplicatorRequestMaxAttempts?: number;
 	keep?: (
 		entry: Entry<Operation<T>> | ShallowEntry | EntryReplicated<R>,
 	) => boolean;
@@ -125,6 +127,10 @@ export class EventStore<
 			trim: properties?.trim,
 			replicas: properties?.replicas,
 			waitForReplicatorTimeout: properties?.waitForReplicatorTimeout,
+			waitForReplicatorRequestIntervalMs:
+				properties?.waitForReplicatorRequestIntervalMs,
+			waitForReplicatorRequestMaxAttempts:
+				properties?.waitForReplicatorRequestMaxAttempts,
 			encoding: JSON_ENCODING,
 			timeUntilRoleMaturity: properties?.timeUntilRoleMaturity ?? 3000,
 			waitForPruneDelay: properties?.waitForPruneDelay ?? 0,

packages/programs/data/shared-log/test/wait-for-replicator.spec.ts

@@ -0,0 +1,52 @@
+import { TestSession } from "@peerbit/test-utils";
+import { TimeoutError } from "@peerbit/time";
+import { expect } from "chai";
+import { RequestReplicationInfoMessage } from "../src/replication.js";
+import { EventStore } from "./utils/stores/index.js";
+
+describe("waitForReplicator", () => {
+	let session: TestSession;
+	let db: EventStore<string, any>;
+
+	afterEach(async () => {
+		if (db && db.closed === false) {
+			await db.drop();
+		}
+		await session?.stop();
+	});
+
+	it("respects configured request retry limits", async () => {
+		session = await TestSession.connected(2);
+		db = await session.peers[0].open(new EventStore<string, any>(), {
+			args: {
+				timeUntilRoleMaturity: 0,
+				waitForReplicatorRequestIntervalMs: 50,
+				waitForReplicatorRequestMaxAttempts: 2,
+			},
+		});
+
+		const originalSend = db.log.rpc.send.bind(db.log.rpc);
+		let requestCount = 0;
+		db.log.rpc.send = async (message: any, options: any) => {
+			if (message instanceof RequestReplicationInfoMessage) {
+				requestCount++;
+				return;
+			}
+			return originalSend(message, options);
+		};
+
+		try {
+			await db.log.waitForReplicator(session.peers[1].identity.publicKey, {
+				timeout: 300,
+				eager: true,
+			});
+			throw new Error("Expected waitForReplicator() to time out");
+		} catch (error) {
+			expect(error).to.be.instanceOf(TimeoutError);
+		} finally {
+			db.log.rpc.send = originalSend;
+		}
+
+		expect(requestCount).to.equal(2);
+	});
+});