fix: stabilize file-share refresh catch-up under churn (#647)

* fix(sync): soften transient entry fetch failures during refresh

* test(blocks): satisfy strict callback typing

* test(blocks): use remote block handler in retry regression

Author: marcus-pousette

packages/log/src/log.ts

@@ -37,6 +37,15 @@ import { Trim, type TrimOptions } from "./trim.js";
 
 const { LastWriteWins } = Sorting;
 
+const isRecoverableJoinResolveError = (error: unknown): boolean => {
+	const message =
+		error instanceof Error ? error.message : typeof error === "string" ? error : "";
+	return (
+		message.includes("Failed to resolve block") ||
+		(error instanceof Error && error.name === "AbortError")
+	);
+};
+
 type CreateSqliteIndexer = typeof import("@peerbit/indexer-sqlite3").create;
 let sqliteCreate: CreateSqliteIndexer | undefined;
 const createDefaultIndexer = async (): Promise<Indices> => {
@@ -846,15 +855,23 @@ export class Log<T> {
 				}
 
 				const from = await options.resolveRemoteFrom?.(a, remote?.signal);
-				const nested =
-					options.references?.get(a) ??
-					(await Entry.fromMultihash<T>(this._storage, a, {
-						remote: {
-							timeout: remote?.timeout,
-							signal: remote?.signal,
-							...(from && from.length > 0 ? { from } : {}),
-						},
-					}));
+				let nested: Entry<T>;
+				try {
+					nested =
+						options.references?.get(a) ??
+						(await Entry.fromMultihash<T>(this._storage, a, {
+							remote: {
+								timeout: remote?.timeout,
+								signal: remote?.signal,
+								...(from && from.length > 0 ? { from } : {}),
+							},
+						}));
+				} catch (error) {
+					if (isRecoverableJoinResolveError(error)) {
+						return false;
+					}
+					throw error;
+				}
 
 				const p = this.joinRecursively(
 					nested,

packages/log/test/join.spec.ts

@@ -1033,6 +1033,30 @@ describe("join", function () {
 				expect(fetchCounter).equal(0); // no fetches since all entries where passed
 				expect(joinEntryCounter).equal(2);
 			});
+
+			it("does not reject join when a nested entry block is temporarily missing", async () => {
+				const { entry: a1 } = await log1.append(new Uint8Array([0, 1]));
+				const { entry: a2 } = await log1.append(new Uint8Array([0, 2]), {
+					meta: { next: [a1] },
+				});
+
+				const fromMultihashStub = sinon
+					.stub(Entry, "fromMultihash")
+					.callsFake(async (store, hash, options) => {
+						if (hash === a1.hash) {
+							throw new Error("Failed to resolve block: " + hash);
+						}
+						return fromMultihashOrg(store, hash, options);
+					});
+
+				try {
+					await log2.join([a2]);
+				} finally {
+					fromMultihashStub.restore();
+				}
+
+				expect(log2.length).equal(0);
+			});
 		});
 
 		// TODO move this into the prune test file

packages/transport/blocks/src/remote.ts

@@ -251,6 +251,19 @@ export class RemoteBlocks implements IBlocks {
 		this._providerCache.add(cidString, normalized);
 	}
 
+	private pickRequestBatch(providers: string[], attempt: number): string[] {
+		if (providers.length <= 1) {
+			return providers;
+		}
+		const batchSize = Math.min(2, providers.length);
+		const start = (attempt * batchSize) % providers.length;
+		const batch: string[] = [];
+		for (let i = 0; i < batchSize; i++) {
+			batch.push(providers[(start + i) % providers.length]);
+		}
+		return this.normalizeProviderHints(batch, batchSize);
+	}
+
 	private async resolveRemoteProviders(
 		cidString: string,
 		options?: { signal?: AbortSignal; refresh?: boolean },
@@ -541,13 +554,20 @@ export class RemoteBlocks implements IBlocks {
 				if (providers.length === 0) return;
 				try {
 					const expiresAt = Date.now() + (options.timeout ?? 30_000);
+					const requestProviders = this.pickRequestBatch(
+						providers,
+						requeryCount,
+					);
 					await this.options.publish(
 						new BlockRequest(cidString),
 						{
 							priority: options.priority,
 							responsePriority: options.priority,
 							expiresAt,
-							mode: new SilentDelivery({ to: providers, redundancy: 1 }),
+							mode: new SilentDelivery({
+								to: requestProviders,
+								redundancy: requestProviders.length,
+							}),
 						},
 					);
 					requeryCount += 1;

packages/transport/blocks/test/libp2p.spec.ts

@@ -233,6 +233,66 @@ describe("transport", function () {
 		expect(new Uint8Array(read!)).to.deep.equal(data);
 	});
 
+	it("probes additional explicit providers when the first candidate does not answer", async () => {
+		session = await TestSession.disconnected(3, {
+			services: { blocks: (c) => new DirectBlock(c) },
+		});
+
+		await store(session, 0).start();
+		await store(session, 1).start();
+		await store(session, 2).start();
+
+		await session.connect([[session.peers[0], session.peers[1]]]);
+		await waitForNeighbour(store(session, 0), store(session, 1));
+
+		const data = new Uint8Array([5, 4, 3]);
+		const cid = await store(session, 0).put(data);
+		expect(cid).equal("zb2rhbnwihVzMMEGAPf9EwTZBsQz9fszCnM4Y8mJmBFgiyN7J");
+
+		const requesterRemoteBlocks = (store(session, 1) as any)[
+			"remoteBlocks"
+		] as RemoteBlocks;
+		const originalPublish = requesterRemoteBlocks.options.publish;
+		let forwardedToSource = 0;
+
+			requesterRemoteBlocks.options.publish = (message: any, options: any) => {
+				if (message instanceof BlockRequest) {
+					const to = (options?.mode as any)?.to ?? [];
+					const redundancy = Math.max(1, (options?.mode as any)?.redundancy ?? 1);
+					const selected = to.slice(0, redundancy);
+					return (async (): Promise<void> => {
+						const sourceRemoteBlocks = (store(session, 0) as any)[
+							"remoteBlocks"
+						] as RemoteBlocks;
+						await Promise.all(
+							selected.map(async (target: string): Promise<void> => {
+								if (target === store(session, 0).publicKeyHash) {
+									forwardedToSource++;
+									await sourceRemoteBlocks.onMessage(message, {
+										from: store(session, 1).publicKeyHash,
+									});
+								}
+							}),
+						);
+					})();
+				}
+				return originalPublish(message, options);
+			};
+
+		try {
+			const read = await store(session, 1).get(cid, {
+				remote: {
+					timeout: 5_000,
+					from: [store(session, 2).publicKeyHash, store(session, 0).publicKeyHash],
+				},
+			});
+			expect(new Uint8Array(read!)).to.deep.equal(data);
+			expect(forwardedToSource).greaterThan(0);
+		} finally {
+			requesterRemoteBlocks.options.publish = originalPublish;
+		}
+	});
+
 	it("only responds to peer that needs block", async () => {
 		session = await TestSession.disconnected(3, {
 			services: { blocks: (c) => new DirectBlock(c) },