Dynamic sharding

.github/workflows/ci.yml

@@ -14,8 +14,8 @@ jobs:
        matrix:
         node-version: [18.x]
         test_cmd: 
-          - yarn playwright install --with-deps && yarn test:node --roots ./packages/clients --w 2 && yarn test:browser
-          - yarn test:node --roots ./packages/programs ./docs ./packages/log  --w 2
+          - yarn playwright install --with-deps && yarn test:node --roots ./docs ./packages/log ./packages/clients --w 2 && yarn test:browser
+          - yarn test:node --roots ./packages/programs --w 2
           - yarn test:node --roots ./packages/transport ./packages/utils --w 2
 
     if: github.event_name == 'push'

packages/clients/peerbit-proxy/interface/src/__tests__/host-client.test.ts

@@ -258,51 +258,86 @@ describe("index", () => {
 			await client1.services.pubsub.waitFor(client2.peerId);
 		});
 
-		it("subscribe/unsubscribe", async () => {
-			await client2.services.pubsub.subscribe("topic");
-			let msg1 = false;
-			let msg1b = false;
-			let msg2 = false;
-			let msg2b = false;
-
-			await client1.services.pubsub.addEventListener("data", () => {
-				msg1 = true;
+		describe("publish", () => {
+			it("multiple hosts", async () => {
+				await client2.services.pubsub.subscribe("topic");
+				let msg1 = false;
+				let msg1b = false;
+				let msg2 = false;
+				let msg2b = false;
+
+				await client1.services.pubsub.addEventListener("data", () => {
+					msg1 = true;
+				});
+
+				await client1b.services.pubsub.addEventListener("data", () => {
+					msg1b = true;
+				});
+
+				await client2.services.pubsub.addEventListener("data", () => {
+					msg2 = true;
+				});
+
+				await client2b.services.pubsub.addEventListener("data", () => {
+					msg2b = true;
+				});
+
+				await client1.services.pubsub.requestSubscribers("topic");
+				await waitForResolved(async () =>
+					expect(
+						(await client1.services.pubsub.getSubscribers("topic"))!.length
+					).toEqual(1)
+				);
+				await client1.services.pubsub.publish(data, { topics: ["topic"] });
+				await waitForResolved(() => expect(msg2).toBeTrue());
+				expect(msg2b).toBeFalse();
+				await client2b.services.pubsub.subscribe("topic");
+				await client1.services.pubsub.publish(data, { topics: ["topic"] });
+				await waitForResolved(() => expect(msg2b).toBeTrue());
+
+				expect(msg1).toBeFalse();
+				expect(msg1b).toBeFalse();
+
+				await client2.services.pubsub.unsubscribe("topic");
+
+				msg2 = false;
+				await client1.services.pubsub.publish(data, { topics: ["topic"] });
+				await delay(3000);
+				expect(msg2).toBeFalse();
 			});
 
-			await client1b.services.pubsub.addEventListener("data", () => {
-				msg1b = true;
-			});
+			it("same host", async () => {
+				let msg1data = false;
+				let msg1publish = false;
+				let msg2data = false;
+				let msg2publish = false;
 
-			await client2.services.pubsub.addEventListener("data", () => {
-				msg2 = true;
-			});
+				await client1.services.pubsub.addEventListener("data", () => {
+					msg1data = true;
+				});
 
-			await client2b.services.pubsub.addEventListener("data", () => {
-				msg2b = true;
-			});
+				await client1.services.pubsub.addEventListener("publish", () => {
+					msg1publish = true;
+				});
 
-			await client1.services.pubsub.requestSubscribers("topic");
-			await waitForResolved(async () =>
-				expect(
-					(await client1.services.pubsub.getSubscribers("topic"))!.length
-				).toEqual(1)
-			);
-			await client1.services.pubsub.publish(data, { topics: ["topic"] });
-			await waitForResolved(() => expect(msg2).toBeTrue());
-			expect(msg2b).toBeFalse();
-			await client2b.services.pubsub.subscribe("topic");
-			await client1.services.pubsub.publish(data, { topics: ["topic"] });
-			await waitForResolved(() => expect(msg2b).toBeTrue());
-
-			expect(msg1).toBeFalse();
-			expect(msg1b).toBeFalse();
-
-			await client2.services.pubsub.unsubscribe("topic");
-
-			msg2 = false;
-			await client1.services.pubsub.publish(data, { topics: ["topic"] });
-			await delay(3000);
-			expect(msg2).toBeFalse();
+				await client1b.services.pubsub.addEventListener("data", () => {
+					msg2data = true;
+				});
+
+				await client1b.services.pubsub.addEventListener("publish", () => {
+					msg2publish = true;
+				});
+
+				await client1.services.pubsub.subscribe("topic");
+				await client1b.services.pubsub.subscribe("topic");
+				await client1.services.pubsub.publish(data, { topics: ["topic"] });
+
+				expect(msg1data).toBeFalse();
+				expect(msg1publish).toBeTrue();
+				await waitForResolved(() => expect(msg2data).toBeTrue());
+
+				expect(msg2publish).toBeTrue(); // TODO expected?
+			});
 		});
 
 		it("getSubscribers", async () => {
@@ -339,10 +374,5 @@ describe("index", () => {
 			await waitForResolved(() => expect(receivedMessages).toHaveLength(1));
 			expect(receivedMessages[0]).toBeInstanceOf(GetSubscribers);
 		});
-
-		it("emitSelf", () => {
-			expect(host1.services.pubsub.emitSelf).toBeFalse();
-			expect(client1.services.pubsub.emitSelf).toBeFalse();
-		});
 	});
 });

packages/clients/peerbit-proxy/interface/src/client.ts

@@ -129,10 +129,8 @@ export class PeerbitProxyClient implements ProgramClient {
 		> = new Map();
 		this._services = {
 			pubsub: {
-				emitSelf: false, // might be changed to true on connect()
 				addEventListener: async (type, lister, options) => {
 					pubsubEventEmitter.addEventListener(type, lister, options);
-
 					let subscription = eventListenerSubscribeCounter.get(type);
 					if (!subscription) {
 						const emitMessageId = randomBytes(32);
@@ -344,6 +342,13 @@ export class PeerbitProxyClient implements ProgramClient {
 					await this.request<memory.MemoryMessage<memory.api.RESP_Open>>(
 						new memory.MemoryMessage(new memory.api.REQ_Open({ level }))
 					);
+				},
+				size: async () => {
+					return (
+						await this.request<memory.MemoryMessage<memory.api.RESP_Size>>(
+							new memory.MemoryMessage(new memory.api.REQ_Size({ level }))
+						)
+					).message.size;
 				}
 			};
 		};
@@ -367,10 +372,6 @@ export class PeerbitProxyClient implements ProgramClient {
 				new network.REQ_GetMultiaddrs()
 			)
 		).multiaddr;
-
-		this.services.pubsub.emitSelf = (
-			await this.request<pubsub.RESP_EmitSelf>(new pubsub.REQ_EmitSelf())
-		).value;
 	}
 
 	getMultiaddrs(): Multiaddr[] {

packages/clients/peerbit-proxy/interface/src/host.ts

@@ -3,7 +3,7 @@ import { PeerId } from "@libp2p/interface/peer-id";
 import { Multiaddr } from "@multiformats/multiaddr";
 import { Blocks } from "@peerbit/blocks-interface";
 import { Keychain, Ed25519Keypair } from "@peerbit/crypto";
-import { DataEvent, PubSub } from "@peerbit/pubsub-interface";
+import { DataEvent, PubSub, PublishEvent } from "@peerbit/pubsub-interface";
 import { ProgramClient } from "@peerbit/program";
 import * as blocks from "./blocks.js";
 import * as keychain from "./keychain.js";
@@ -14,7 +14,7 @@ import { Message } from "./message.js";
 import * as network from "./network.js";
 import * as pubsub from "./pubsub.js";
 import * as connection from "./connection.js";
-
+import { CustomEvent } from "@libp2p/interface/events";
 import { serialize, deserialize } from "@dao-xyz/borsh";
 
 const levelKey = (level: string[]) => JSON.stringify(level);
@@ -43,6 +43,20 @@ export class PeerbitProxyHost implements ProgramClient {
 		this._pubsubTopicSubscriptions = new Map();
 		this._memoryIterator = new Map();
 
+		const dispatchFunction = this.hostClient.services.pubsub.dispatchEvent.bind(
+			this.hostClient.services.pubsub
+		);
+
+		// Override pubsub dispatchEvent so that data that is published from one client
+		// appears in other clients as incoming data messages.
+		// this allows multiple clients to subscribe to share the same host and
+		// also have same databases open
+		this.hostClient.services.pubsub.dispatchEvent = (evt: CustomEvent<any>) => {
+			if (evt.type === "publish") {
+				dispatchFunction(new CustomEvent("data", { detail: evt.detail }));
+			}
+			return dispatchFunction(evt);
+		};
 		this.messages.start();
 	}
 
@@ -308,43 +322,54 @@ export class PeerbitProxyHost implements ProgramClient {
 				}
 				let subscription = map.get(message.type);
 				if (!subscription) {
-					subscription = {
-						counter: 1,
-						fn: async (e) => {
-							// TODO what if many clients whants the same data, dedup serialization invokations?
-							if (e.detail instanceof DataEvent) {
-								const subscriptions = this._pubsubTopicSubscriptions.get(
-									from.id
-								);
-								let found = false;
-								if (subscriptions) {
-									for (const topic of e.detail.data.topics) {
-										found = subscriptions.has(topic);
-										if (found) {
-											break;
-										}
-									}
-								}
+					const cb = async (e: CustomEvent<any>) => {
+						// TODO what if many clients whants the same data, dedup serialization invokations?
+						if (
+							e.detail instanceof PublishEvent &&
+							e.detail.client === from.id &&
+							message.type === "data"
+						) {
+							// ignore 'publish' events routed to 'data' events if the dispatcher is the same as the receiver
+							return;
+						}
 
-								if (!found) {
-									// Ignore this message, since the client is not subscribing to any of the topics
-									return;
+						if (
+							e.detail instanceof DataEvent ||
+							e.detail instanceof PublishEvent
+						) {
+							const subscriptions = this._pubsubTopicSubscriptions.get(from.id);
+							let found = false;
+							if (subscriptions) {
+								for (const topic of e.detail.data.topics) {
+									found = subscriptions.has(topic);
+									if (found) {
+										break;
+									}
 								}
 							}
 
-							const request = new pubsub.RESP_EmitEvent(
-								message.type,
-								serialize(e.detail)
-							);
-							request.messageId = message.emitMessageId; // Same message id so that receiver can subscribe to all events emitted from this listener
-							await this.messages.send(serialize(request), from.id);
+							if (!found) {
+								// Ignore this message, since the client is not subscribing to any of the topics
+								return;
+							}
 						}
+
+						const request = new pubsub.RESP_EmitEvent(
+							message.type,
+							serialize(e.detail)
+						);
+						request.messageId = message.emitMessageId; // Same message id so that receiver can subscribe to all events emitted from this listener
+						await this.messages.send(serialize(request), from.id);
 					};
+					subscription = {
+						counter: 1,
+						fn: cb
+					};
+
 					await this.services.pubsub.addEventListener(
 						message.type,
 						subscription.fn
 					);
-
 					map.set(message.type, subscription);
 				} else {
 					subscription.counter += 1;
@@ -378,11 +403,22 @@ export class PeerbitProxyHost implements ProgramClient {
 					message.type,
 					message.data
 				);
+				const dispatched =
+					await this.services.pubsub.dispatchEvent(customEvent);
+
+				/* 	
+				if (message.type === 'publish') {
+						await this.services.pubsub.dispatchEvent(
+							pubsub.createCustomEventFromType(
+								'data',
+								message.data
+							)
+						)
+					}
+				 */
 				await this.respond(
 					message,
-					new pubsub.RESP_DispatchEvent(
-						await this.services.pubsub.dispatchEvent(customEvent)
-					),
+					new pubsub.RESP_DispatchEvent(dispatched),
 					from
 				);
 			} else if (message instanceof pubsub.REQ_GetSubscribers) {
@@ -398,9 +434,9 @@ export class PeerbitProxyHost implements ProgramClient {
 					message,
 					new pubsub.RESP_Publish(
 						await this.services.pubsub.publish(message.data, {
-							strict: message.strict,
 							to: message.to!,
-							topics: message.topics!
+							topics: message.topics!,
+							client: from.id
 						})
 					),
 					from
@@ -435,12 +471,6 @@ export class PeerbitProxyHost implements ProgramClient {
 					),
 					from
 				);
-			} else if (message instanceof pubsub.REQ_EmitSelf) {
-				await this.respond(
-					message,
-					new pubsub.RESP_EmitSelf(this.services.pubsub.emitSelf),
-					from
-				);
 			} else {
 				throw new Error("Unknown message type: " + message.constructor.name);
 			}