fix: allowing searching many documents

packages/programs/data/document/src/document-store.ts

@@ -32,10 +32,12 @@ import {
 	PutOperation,
 	CanSearch,
 	CanRead,
-	IndexedDB
+	InMemoryIndex,
+	MAX_DOCUMENT_SIZE
 } from "./document-index.js";
 import { asString, checkKeyable, Keyable } from "./utils.js";
 import { Context, Results } from "./query.js";
+export { MAX_DOCUMENT_SIZE };
 
 const logger = loggerFn({ module: "document" });
 
@@ -81,7 +83,7 @@ export type SetupOptions<T> = {
 @variant("documents")
 export class Documents<T extends Record<string, any>>
 	extends Program<SetupOptions<T>, DocumentEvents<T> & ProgramEvents>
-	implements IndexedDB<T>
+	implements InMemoryIndex<T>
 {
 	@field({ type: SharedLog })
 	log: SharedLog<Operation<T>>;
@@ -311,6 +313,14 @@ export class Documents<T extends Record<string, any>>
 		const key = this._index.indexByResolver(doc as any as Keyable);
 		checkKeyable(key);
 		const ser = serialize(doc);
+		if (ser.length > MAX_DOCUMENT_SIZE) {
+			throw new Error(
+				`Document is too large (${
+					ser.length * 1e-6
+				}) mb). Needs to be less than ${MAX_DOCUMENT_SIZE * 1e-6} mb`
+			);
+		}
+
 		const existingDocument = options?.unique
 			? undefined
 			: (
@@ -449,7 +459,7 @@ export class Documents<T extends Record<string, any>>
 						context,
 						reference:
 							valueToIndex === value || value instanceof Program
-								? value
+								? { value, last: payload }
 								: undefined
 					});
 				} else if (

packages/programs/data/shared-log/src/__tests__/replicate.test.ts

@@ -313,6 +313,27 @@ describe(`exchange`, function () {
 			}
 		}
 	});
+	it("replicates database of large entries", async () => {
+		let count = 10;
+		for (let i = 0; i < count; i++) {
+			const value = toBase64(randomBytes(4e6));
+			await db1.add(value, { meta: { next: [] } }); // force unique heads
+		}
+		db2 = (await EventStore.open<EventStore<string>>(
+			db1.address!,
+			session.peers[1],
+			{
+				args: {
+					role: {
+						type: "replicator",
+						factor: 1
+					}
+				}
+			}
+		))!;
+
+		await waitForResolved(() => expect(db2.log.log.length).toEqual(count));
+	});
 
 	describe("redundancy", () => {
 		it("only sends entries once", async () => {

packages/programs/data/shared-log/src/exchange-heads.ts

@@ -112,6 +112,7 @@ export const createExchangeHeadsMessages = async (
 
 		size += fromHead.size;
 		if (size > MAX_EXCHANGE_MESSAGE_SIZE) {
+			size = 0;
 			messages.push(
 				new ExchangeHeadsMessage({
 					heads: current

packages/transport/pubsub/src/index.ts

@@ -38,7 +38,12 @@ import { getPublicKeyFromPeerId, PublicSignKey } from "@peerbit/crypto";
 import { CustomEvent } from "@libp2p/interface";
 
 export const logger = logFn({ module: "lazysub", level: "warn" });
-const logError = (e?: { message: string }) => logger.error(e?.message);
+const logError = (e?: { message: string }) => {
+	logger.error(e?.message);
+};
+const logErrorIfStarted = (e?: { message: string }) => {
+	e instanceof NotStartedError === false && logError(e);
+};
 
 export interface PeerStreamsInit {
 	id: Libp2pPeerId;
@@ -553,7 +558,7 @@ export class DirectSub extends DirectStream<PubSubEvents> implements PubSub {
 				message.header.mode instanceof SeekDelivery
 			) {
 				// DONT await this since it might introduce a dead-lock
-				this.relayMessage(from, message).catch(logError);
+				this.relayMessage(from, message).catch(logErrorIfStarted);
 			}
 		} else {
 			if ((await this.verifyAndProcess(message)) === false) {
@@ -657,7 +662,7 @@ export class DirectSub extends DirectStream<PubSubEvents> implements PubSub {
 
 				// Forward
 				// DONT await this since it might introduce a dead-lock
-				this.relayMessage(from, message).catch(logError);
+				this.relayMessage(from, message).catch(logErrorIfStarted);
 			} else if (pubsubMessage instanceof Unsubscribe) {
 				if (this.subscriptionMessageIsLatest(message, pubsubMessage)) {
 					const changed: string[] = [];
@@ -693,7 +698,7 @@ export class DirectSub extends DirectStream<PubSubEvents> implements PubSub {
 				}
 
 				// DONT await this since it might introduce a dead-lock
-				this.relayMessage(from, message).catch(logError);
+				this.relayMessage(from, message).catch(logErrorIfStarted);
 			} else if (pubsubMessage instanceof GetSubscribers) {
 				const subscriptionsToSend: string[] = this.getSubscriptionOverlap(
 					pubsubMessage.topics
@@ -723,7 +728,7 @@ export class DirectSub extends DirectStream<PubSubEvents> implements PubSub {
 
 				// Forward
 				// DONT await this since it might introduce a dead-lock
-				this.relayMessage(from, message).catch(logError);
+				this.relayMessage(from, message).catch(logErrorIfStarted);
 			}
 		}
 		return true;

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

@@ -24,4 +24,8 @@ export interface WaitForPeer {
 	): Promise<void>;
 }
 
-export class NotStartedError extends Error {}
+export class NotStartedError extends Error {
+	constructor() {
+		super("Not started");
+	}
+}

packages/transport/stream/src/index.ts

@@ -62,6 +62,7 @@ import {
 
 import { MultiAddrinfo } from "@peerbit/stream-interface";
 import { BandwidthTracker } from "./stats.js";
+export { BandwidthTracker }; // might be useful for others
 
 const logError = (e?: { message: string }) => {
 	return logger.error(e?.message);
@@ -618,35 +619,35 @@ export abstract class DirectStream<
 			return;
 		}
 
-		// reset and clear up
-
-		this.started = false;
-
-		this.closeController.abort();
-
 		clearTimeout(this.pruneConnectionsTimeout);
 
 		await Promise.all(
 			this.multicodecs.map((x) => this.components.registrar.unhandle(x))
 		);
 
+		// unregister protocol and handlers
+		if (this._registrarTopologyIds != null) {
+			this._registrarTopologyIds?.map((id) =>
+				this.components.registrar.unregister(id)
+			);
+		}
+
+		// reset and clear up
+		this.started = false;
+
+		this.closeController.abort();
+
 		logger.debug("stopping");
 		for (const peerStreams of this.peers.values()) {
 			await peerStreams.close();
 		}
+
 		for (const [k, v] of this.healthChecks) {
 			clearTimeout(v);
 		}
 		this.healthChecks.clear();
 		this.prunedConnectionsCache?.clear();
 
-		// unregister protocol and handlers
-		if (this._registrarTopologyIds != null) {
-			this._registrarTopologyIds?.map((id) =>
-				this.components.registrar.unregister(id)
-			);
-		}
-
 		this.queue.clear();
 		this.peers.clear();
 		this.seenCache.clear();
@@ -1662,8 +1663,8 @@ export abstract class DirectStream<
 		to?: PeerStreams[] | Map<string, PeerStreams>,
 		relayed?: boolean
 	): Promise<void> {
-		if (this.stopping) {
-			return;
+		if (this.stopping || !this.started) {
+			throw new NotStartedError();
 		}
 
 		let delivereyPromise: Promise<void> | undefined = undefined as any;