shared-log: chunk repair sync and preserve rateless sweep batches

Author: marcus-pousette

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

@@ -24,6 +24,23 @@ export type SyncOptions<R extends "u32" | "u64"> = {
 	 * high-priority entries using the simple synchronizer.
 	 */
 	maxSimpleEntries?: number;
+
+	/**
+	 * Maximum number of hash strings in one simple sync message.
+	 */
+	maxSimpleHashesPerMessage?: number;
+
+	/**
+	 * Maximum number of coordinates in one simple sync coordinate message.
+	 */
+	maxSimpleCoordinatesPerMessage?: number;
+
+	/**
+	 * Maximum number of hashes tracked per convergent repair session target.
+	 * Large sessions still dispatch all entries, but only this many are tracked
+	 * for deterministic completion metadata.
+	 */
+	maxConvergentTrackedHashes?: number;
 };
 
 export type SynchronizerComponents<R extends "u32" | "u64"> = {
@@ -51,6 +68,8 @@ export type RepairSessionResult = {
 	attempts: number;
 	durationMs: number;
 	completed: boolean;
+	requestedTotal?: number;
+	truncated?: boolean;
 };
 
 export type RepairSession = {

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

@@ -58,6 +58,10 @@ const getSyncIdString = (message: { syncId: Uint8Array }) => {
 	return toBase64(message.syncId);
 };
 
+const DEFAULT_CONVERGENT_REPAIR_TIMEOUT_MS = 30_000;
+const DEFAULT_CONVERGENT_RETRY_INTERVALS_MS = [0, 1_000, 3_000, 7_000];
+const DEFAULT_MAX_CONVERGENT_TRACKED_HASHES = 4_096;
+
 @variant([3, 0])
 export class StartSync extends TransportMessage {
 	@field({ type: Uint8Array })
@@ -280,6 +284,46 @@ export class RatelessIBLTSynchronizer<D extends "u32" | "u64">
 		this.startedOrCompletedSynchronizations = new Cache({ max: 1e4 });
 	}
 
+	private get maxConvergentTrackedHashes() {
+		const value = this.properties.sync?.maxConvergentTrackedHashes;
+		return value && Number.isFinite(value) && value > 0
+			? Math.floor(value)
+			: DEFAULT_MAX_CONVERGENT_TRACKED_HASHES;
+	}
+
+	private normalizeRetryIntervals(retryIntervalsMs?: number[]): number[] {
+		if (!retryIntervalsMs || retryIntervalsMs.length === 0) {
+			return [...DEFAULT_CONVERGENT_RETRY_INTERVALS_MS];
+		}
+
+		return [...retryIntervalsMs]
+			.map((x) => Math.max(0, Math.floor(x)))
+			.filter((x, i, arr) => arr.indexOf(x) === i)
+			.sort((a, b) => a - b);
+	}
+
+	private getPrioritizedEntries(entries: Map<string, EntryReplicated<D>>) {
+		const priorityFn = this.properties.sync?.priority;
+		if (!priorityFn) {
+			return [...entries.values()];
+		}
+
+		let index = 0;
+		const scored: { entry: EntryReplicated<D>; index: number; priority: number }[] =
+			[];
+		for (const entry of entries.values()) {
+			const priorityValue = priorityFn(entry);
+			scored.push({
+				entry,
+				index,
+				priority: Number.isFinite(priorityValue) ? priorityValue : 0,
+			});
+			index += 1;
+		}
+		scored.sort((a, b) => b.priority - a.priority || a.index - b.index);
+		return scored.map((x) => x.entry);
+	}
+
 	startRepairSession(properties: {
 		entries: Map<string, EntryReplicated<D>>;
 		targets: string[];
@@ -288,17 +332,98 @@ export class RatelessIBLTSynchronizer<D extends "u32" | "u64">
 		retryIntervalsMs?: number[];
 	}): RepairSession {
 		const mode = properties.mode ?? "best-effort";
+		const targets = [...new Set(properties.targets)];
+		const timeoutMs = Math.max(
+			1,
+			Math.floor(properties.timeoutMs ?? DEFAULT_CONVERGENT_REPAIR_TIMEOUT_MS),
+		);
+		const retryIntervalsMs = this.normalizeRetryIntervals(properties.retryIntervalsMs);
+		const trackedLimit = this.maxConvergentTrackedHashes;
+		const requestedHashes = [...properties.entries.keys()];
+		const requestedHashesTracked = requestedHashes.slice(0, trackedLimit);
+		const truncated = requestedHashesTracked.length < requestedHashes.length;
+
 		if (mode === "convergent") {
-			return this.simple.startRepairSession({
-				...properties,
+			if (properties.entries.size <= trackedLimit) {
+				return this.simple.startRepairSession({
+					...properties,
+					mode: "convergent",
+					timeoutMs,
+					retryIntervalsMs,
+				});
+			}
+
+			const id = `rateless-repair-${++this.repairSessionCounter}`;
+			const startedAt = Date.now();
+			const prioritized = this.getPrioritizedEntries(properties.entries);
+			const trackedEntries = new Map<string, EntryReplicated<D>>();
+			for (const entry of prioritized.slice(0, trackedLimit)) {
+				trackedEntries.set(entry.hash, entry);
+			}
+
+			let cancelled = false;
+			const trackedSession = this.simple.startRepairSession({
+				entries: trackedEntries,
+				targets,
 				mode: "convergent",
+				timeoutMs,
+				retryIntervalsMs,
 			});
+
+			const runDispatchSchedule = async () => {
+				let previousDelay = 0;
+				for (const delayMs of retryIntervalsMs) {
+					if (cancelled) {
+						return;
+					}
+					const elapsed = Date.now() - startedAt;
+					if (elapsed >= timeoutMs) {
+						return;
+					}
+					const waitMs = Math.max(0, delayMs - previousDelay);
+					previousDelay = delayMs;
+					if (waitMs > 0) {
+						await new Promise<void>((resolve) => {
+							const timer = setTimeout(resolve, waitMs);
+							timer.unref?.();
+						});
+					}
+					if (cancelled) {
+						return;
+					}
+					try {
+						await this.onMaybeMissingEntries({
+							entries: properties.entries,
+							targets,
+						});
+					} catch {
+						// Best-effort schedule: tracked session timeout/result decides completion.
+					}
+				}
+			};
+
+			const done = (async (): Promise<RepairSessionResult[]> => {
+				await runDispatchSchedule();
+				const trackedResults = await trackedSession.done;
+				return trackedResults.map((result) => ({
+					...result,
+					requestedTotal: requestedHashes.length,
+					truncated: true,
+				}));
+			})();
+
+			return {
+				id,
+				done,
+				cancel: () => {
+					cancelled = true;
+					trackedSession.cancel();
+				},
+			};
 		}
 
 		const id = `rateless-repair-${++this.repairSessionCounter}`;
 		const startedAt = Date.now();
-		const requestedHashes = [...properties.entries.keys()];
-		const targets = [...new Set(properties.targets)];
 		const done = (async (): Promise<RepairSessionResult[]> => {
 			await this.onMaybeMissingEntries({
 				entries: properties.entries,
@@ -307,12 +432,14 @@ export class RatelessIBLTSynchronizer<D extends "u32" | "u64">
 			const durationMs = Date.now() - startedAt;
 			return targets.map((target) => ({
 				target,
-				requested: requestedHashes.length,
+				requested: requestedHashesTracked.length,
 				resolved: 0,
-				unresolved: [...requestedHashes],
+				unresolved: [...requestedHashesTracked],
 				attempts: 1,
 				durationMs,
 				completed: false,
+				requestedTotal: requestedHashes.length,
+				truncated,
 			}));
 		})();
 		return {