feat(transport): unify directstream and fanout route hints

Author: marcus-pousette

packages/transport/pubsub-interface/src/index.ts

@@ -9,6 +9,7 @@ import {
 	type PeerEvents,
 	type PriorityOptions,
 	type PublicKeyFromHashResolver,
+	type RouteHint,
 	type WaitForPeer,
 	type WithExtraSigners,
 } from "@peerbit/stream-interface";
@@ -136,6 +137,10 @@ export interface PubSub
 		WaitForPeer,
 		PublicKeyFromHashResolver {
 	getSubscribers(topic: string): MaybePromise<PublicSignKey[] | undefined>;
+	getUnifiedRouteHints?(
+		topic: string,
+		targetHash: string,
+	): MaybePromise<RouteHint[]>;
 
 	requestSubscribers(topic: string, from?: PublicSignKey): MaybePromise<void>;
 

packages/transport/pubsub/src/fanout-tree.ts

@@ -8,7 +8,12 @@ import {
 	type PeerStreams,
 	dontThrowIfDeliveryError,
 } from "@peerbit/stream";
-import { AnyWhere, DataMessage, type StreamEvents } from "@peerbit/stream-interface";
+import {
+	AnyWhere,
+	DataMessage,
+	type FanoutRouteTokenHint,
+	type StreamEvents,
+} from "@peerbit/stream-interface";
 import { AbortError, TimeoutError, delay } from "@peerbit/time";
 import { anySignal } from "any-signal";
 import { Uint8ArrayList } from "uint8arraylist";
@@ -2400,6 +2405,50 @@ export class FanoutTree extends DirectStream<FanoutTreeEvents> {
 		return [...ch.routeFromRoot];
 	}
 
+	public getRouteHint(
+		topic: string,
+		root: string,
+		targetHash: string,
+	): FanoutRouteTokenHint | undefined {
+		const id = this.getChannelId(topic, root);
+		const ch = this.channelsBySuffixKey.get(id.suffixKey);
+		if (!ch) return undefined;
+
+		if (targetHash === this.publicKeyHash && ch.routeFromRoot?.length) {
+			return {
+				kind: "fanout-token",
+				root,
+				target: targetHash,
+				route: [...ch.routeFromRoot],
+				updatedAt: Date.now(),
+			};
+		}
+		if (ch.isRoot && ch.children.has(targetHash)) {
+			return {
+				kind: "fanout-token",
+				root,
+				target: targetHash,
+				route: [root, targetHash],
+				updatedAt: Date.now(),
+			};
+		}
+
+		const route = this.getCachedRoute(ch, targetHash);
+		if (!route) return undefined;
+		const entry = ch.routeByPeer.get(targetHash);
+		const updatedAt = entry?.updatedAt ?? Date.now();
+
+		return {
+			kind: "fanout-token",
+			root,
+			target: targetHash,
+			route,
+			updatedAt,
+			expiresAt:
+				ch.routeCacheTtlMs > 0 ? updatedAt + ch.routeCacheTtlMs : undefined,
+		};
+	}
+
 	private cacheRoute(ch: ChannelState, route: string[]) {
 		if (!route?.length) return;
 		if (route[0] !== ch.id.root) return;

packages/transport/pubsub/src/index.ts

@@ -36,6 +36,7 @@ import {
 	MessageHeader,
 	NotStartedError,
 	type PriorityOptions,
+	type RouteHint,
 	SilentDelivery,
 	type WithExtraSigners,
 	deliveryModeHasReceiver,
@@ -1280,6 +1281,39 @@ export class TopicControlPlane
 		return ret;
 	}
 
+	/**
+	 * Returns best-effort route hints for a target peer by combining:
+	 * - DirectStream ACK-learned routes
+	 * - Fanout route tokens for the topic's shard overlay
+	 */
+	getUnifiedRouteHints(topic: string, targetHash: string): RouteHint[] {
+		const hints: RouteHint[] = [];
+		const directHint = this.getBestRouteHint(targetHash);
+		if (directHint) {
+			hints.push(directHint);
+		}
+
+		const topicString = topic.toString();
+		const shardTopic = topicString.startsWith(this.shardTopicPrefix)
+			? topicString
+			: this.getShardTopicForUserTopic(topicString);
+		const shard = this.fanoutChannels.get(shardTopic);
+		if (!shard) {
+			return hints;
+		}
+
+		const fanoutHint = this.fanout.getRouteHint(
+			shardTopic,
+			shard.root,
+			targetHash,
+		);
+		if (fanoutHint) {
+			hints.push(fanoutHint);
+		}
+
+		return hints;
+	}
+
 	async requestSubscribers(
 		topic: string | string[],
 		to?: PublicSignKey,

packages/transport/pubsub/test/fanout-topics.spec.ts

@@ -177,6 +177,31 @@ describe("pubsub (fanout topics)", function () {
 		}
 	});
 
+	it("exposes unified route hints from directstream and fanout", async () => {
+		const { session, configureBootstraps, configureShards } = await createSession(2);
+
+		try {
+			const TOPIC = "fanout-route-hints-topic";
+			await configureShards([0]);
+			configureBootstraps([0]);
+
+			await session.peers[0]!.services.pubsub.subscribe(TOPIC);
+			await session.peers[1]!.services.pubsub.subscribe(TOPIC);
+
+			const targetHash = session.peers[1]!.services.pubsub.publicKeyHash;
+			await waitForResolved(() => {
+				const hints = session.peers[0]!.services.pubsub.getUnifiedRouteHints!(
+					TOPIC,
+					targetHash,
+				);
+				expect(hints.some((h) => h.kind === "directstream-ack")).to.equal(true);
+				expect(hints.some((h) => h.kind === "fanout-token")).to.equal(true);
+			});
+		} finally {
+			await session.stop();
+		}
+	});
+
 	it("preserves publish id/priority/signatures for fanout-backed topics", async () => {
 		const { session, configureBootstraps, configureShards } = await createSession(3);
 

packages/transport/stream-interface/src/index.ts

@@ -18,6 +18,7 @@ export interface StreamEvents extends PeerEvents, MessageEvents {
 }
 
 export * from "./messages.js";
+export * from "./route-hints.js";
 
 // ---------- wait for peer types ----------
 export type Target = "neighbor" | "reachable";

packages/transport/stream-interface/src/route-hints.ts

@@ -0,0 +1,21 @@
+export type DirectStreamAckRouteHint = {
+	kind: "directstream-ack";
+	from: string;
+	target: string;
+	nextHop: string;
+	distance: number;
+	session: number;
+	updatedAt: number;
+	expiresAt?: number;
+};
+
+export type FanoutRouteTokenHint = {
+	kind: "fanout-token";
+	root: string;
+	target: string;
+	route: string[];
+	updatedAt: number;
+	expiresAt?: number;
+};
+
+export type RouteHint = DirectStreamAckRouteHint | FanoutRouteTokenHint;

packages/transport/stream/src/index.ts

@@ -47,6 +47,7 @@ import {
 	getMsgId,
 } from "@peerbit/stream-interface";
 import type {
+	DirectStreamAckRouteHint,
 	IdOptions,
 	PeerRefs,
 	PriorityOptions,
@@ -1748,6 +1749,20 @@ export abstract class DirectStream<
 		this.routes.updateSession(key, undefined);
 	}
 
+	public getRouteHints(
+		target: string,
+		from: string = this.publicKeyHash,
+	): DirectStreamAckRouteHint[] {
+		return this.routes.getRouteHints(from, target);
+	}
+
+	public getBestRouteHint(
+		target: string,
+		from: string = this.publicKeyHash,
+	): DirectStreamAckRouteHint | undefined {
+		return this.routes.getBestRouteHint(from, target);
+	}
+
 	public onPeerSession(key: PublicSignKey, session: number) {
 		this.dispatchEvent(
 			// TODO types

packages/transport/stream/src/routes.ts

@@ -1,3 +1,5 @@
+import type { DirectStreamAckRouteHint } from "@peerbit/stream-interface";
+
 export const MAX_ROUTE_DISTANCE = Number.MAX_SAFE_INTEGER - 1;
 
 const DEFAULT_MAX_FROM_ENTRIES = 2048;
@@ -7,6 +9,7 @@ const DEFAULT_MAX_RELAYS_PER_TARGET = 32;
 type RelayInfo = {
 	session: number;
 	hash: string;
+	updatedAt: number;
 	expireAt?: number;
 	distance: number;
 };
@@ -257,6 +260,7 @@ export class Routes {
 					if (route.distance > distance) {
 						route.distance = distance;
 						route.session = session;
+						route.updatedAt = +new Date();
 						route.expireAt = undefined; // remove expiry since we updated
 						sortRoutes(prev.list);
 						if (prev.list.length > this.maxRelaysPerTarget) {
@@ -267,6 +271,7 @@ export class Routes {
 						return isNewRemoteSession ? "restart" : "updated";
 					} else if (route.distance === distance) {
 						route.session = session;
+						route.updatedAt = +new Date();
 						route.expireAt = undefined; // remove expiry since we updated
 						if (prev.list.length > this.maxRelaysPerTarget) {
 							prev.list.length = this.maxRelaysPerTarget;
@@ -290,6 +295,7 @@ export class Routes {
 				distance,
 				session,
 				hash: neighbour,
+				updatedAt: +new Date(),
 				expireAt: isOldSession
 					? +new Date() + this.routeMaxRetentionPeriod
 					: undefined,
@@ -364,6 +370,38 @@ export class Routes {
 		return this.routes.get(from)?.get(target);
 	}
 
+	getRouteHints(from: string, target: string): DirectStreamAckRouteHint[] {
+		const route = this.routes.get(from)?.get(target);
+		if (!route) {
+			return [];
+		}
+		const now = Date.now();
+		const out: DirectStreamAckRouteHint[] = [];
+		for (const next of route.list) {
+			if (next.expireAt != null && next.expireAt < now) {
+				continue;
+			}
+			out.push({
+				kind: "directstream-ack",
+				from,
+				target,
+				nextHop: next.hash,
+				distance: next.distance,
+				session: next.session,
+				updatedAt: next.updatedAt,
+				expiresAt: next.expireAt,
+			});
+		}
+		return out;
+	}
+
+	getBestRouteHint(
+		from: string,
+		target: string,
+	): DirectStreamAckRouteHint | undefined {
+		return this.getRouteHints(from, target)[0];
+	}
+
 	isReachable(from: string, target: string, maxDistance = MAX_ROUTE_DISTANCE) {
 		const remoteInfo = this.remoteInfo.get(target);
 		const routeInfo = this.routes.get(from)?.get(target);