From 3e58b8654076755251da40685d6e1ae14ebced79 Mon Sep 17 00:00:00 2001 From: xash Date: Thu, 21 Nov 2024 07:41:21 +0100 Subject: [PATCH 1/3] Add range counter tracking Fixes also a bug where a received ReconciliationAnnounceEntries message with wantResponse = true did not increase your_range_counter. --- src/wgps/range_counter.ts | 27 ++++ src/wgps/wgps_messenger.test.ts | 276 ++++++++++++++++++++++++++++++++ src/wgps/wgps_messenger.ts | 24 ++- 3 files changed, 322 insertions(+), 5 deletions(-) create mode 100644 src/wgps/range_counter.ts diff --git a/src/wgps/range_counter.ts b/src/wgps/range_counter.ts new file mode 100644 index 0000000..14e4e8d --- /dev/null +++ b/src/wgps/range_counter.ts @@ -0,0 +1,27 @@ +import { WillowError } from "../errors.ts"; +import { COVERS_NONE } from "./types.ts"; + +export class RangeCounter { + private count = 0n; + private inProgress = new Set(); + + getNext() { + this.inProgress.add(this.count); + return this.count++; + } + + done(covers: bigint | typeof COVERS_NONE) { + if (covers !== COVERS_NONE) { + if (!this.inProgress.delete(covers)) { + throw new WillowError("Answered range without request"); + } + } + } + + info() { + return { + remaining: BigInt(this.inProgress.size), + all: this.count, + }; + } +} diff --git a/src/wgps/wgps_messenger.test.ts b/src/wgps/wgps_messenger.test.ts index b7a7d28..d3cb3d1 100644 --- a/src/wgps/wgps_messenger.test.ts +++ b/src/wgps/wgps_messenger.test.ts @@ -2653,6 +2653,282 @@ function testWgpsMessenger(scenario: WgpsScenario) { }, ); + Deno.test( + `Full Reconciliation with a lot of branching (${scenario.name})`, + async (test) => { + await test.step("sync", async () => { + const [alfie, betty] = transportPairInMemory(); + + const challengeHash = async (bytes: Uint8Array) => { + return new Uint8Array(await crypto.subtle.digest("SHA-256", bytes)); + }; + + const storeMapAlfie = new StoreMap( + async (namespace) => { + const drivers = await scenario.getDrivers("alfie", namespace); + + const store = new Store({ + namespace, + schemes: { + authorisation: testSchemeAuthorisation, + fingerprint: testSchemeFingerprint, + namespace: testSchemeNamespace, + path: testSchemePath, + payload: testSchemePayload, + subspace: testSchemeSubspace, + }, + ...drivers, + }); + + return store; + }, + { + authorisation: testSchemeAuthorisation, + fingerprint: testSchemeFingerprint, + namespace: testSchemeNamespace, + path: testSchemePath, + payload: testSchemePayload, + subspace: testSchemeSubspace, + }, + ); + + const storeFamilyAlfie = await storeMapAlfie.get(TestNamespace.Family); + + const storeMapBetty = new StoreMap( + async (namespace) => { + const drivers = await scenario.getDrivers("betty", namespace); + const store = new Store({ + namespace, + schemes: { + authorisation: testSchemeAuthorisation, + fingerprint: testSchemeFingerprint, + namespace: testSchemeNamespace, + path: testSchemePath, + payload: testSchemePayload, + subspace: testSchemeSubspace, + }, + ...drivers, + }); + + return store; + }, + { + authorisation: testSchemeAuthorisation, + fingerprint: testSchemeFingerprint, + namespace: testSchemeNamespace, + path: testSchemePath, + payload: testSchemePayload, + subspace: testSchemeSubspace, + }, + ); + + const storeFamilyBetty = await storeMapBetty.get(TestNamespace.Family); + + for (let i = 0; i < 100; i++) { + const payload = new Uint8Array(2 ** 4); + payload[0] = i; + const entry = { + subspace: TestSubspace.Gemma, + payload: payload, + path: [ + new Uint8Array([1]), + crypto.getRandomValues(new Uint8Array(4)), + ], + }; + if (i % 2) { + await storeFamilyAlfie.set(entry, TestSubspace.Gemma); + } else { + await storeFamilyBetty.set(entry, TestSubspace.Gemma); + } + } + + const messengerAlfie = new WgpsMessenger({ + challengeHash, + challengeLength: 128, + challengeHashLength: 32, + maxPayloadSizePower: 8, + transport: alfie, + schemes: { + subspaceCap: testSchemeSubspaceCap, + namespace: testSchemeNamespace, + accessControl: testSchemeAccessControl, + pai: testSchemePai, + path: testSchemePath, + subspace: testSchemeSubspace, + authorisationToken: testSchemeAuthorisationToken, + fingerprint: testSchemeFingerprint, + authorisation: testSchemeAuthorisation, + payload: testSchemePayload, + }, + getStore: (namespace) => { + return storeMapAlfie.get(namespace); + }, + interests: new Map([[ + { + capability: { + namespace: TestNamespace.Family, + subspace: TestSubspace.Gemma, + path: [new Uint8Array([1])], + receiver: TestSubspace.Alfie, + time: { + start: BigInt(0), + end: OPEN_END, + }, + } as TestReadCap, + }, + [{ + area: { + includedSubspaceId: TestSubspace.Gemma, + pathPrefix: [new Uint8Array([1])], + timeRange: { + start: BigInt(0), + end: OPEN_END, + }, + }, + maxCount: 0, + maxSize: BigInt(0), + }], + ]]), + transformPayload: (bytes) => bytes, + processReceivedPayload: (bytes) => bytes, + }); + + const messengerBetty = new WgpsMessenger({ + challengeHash, + challengeLength: 128, + challengeHashLength: 32, + maxPayloadSizePower: 8, + transport: betty, + schemes: { + subspaceCap: testSchemeSubspaceCap, + namespace: testSchemeNamespace, + accessControl: testSchemeAccessControl, + pai: testSchemePai, + path: testSchemePath, + subspace: testSchemeSubspace, + authorisationToken: testSchemeAuthorisationToken, + fingerprint: testSchemeFingerprint, + authorisation: testSchemeAuthorisation, + payload: testSchemePayload, + }, + getStore: (namespace) => { + return storeMapBetty.get(namespace); + }, + interests: new Map([[ + { + capability: { + namespace: TestNamespace.Family, + subspace: TestSubspace.Gemma, + path: [new Uint8Array([1])], + receiver: TestSubspace.Betty, + time: { + start: BigInt(0), + end: OPEN_END, + }, + } as TestReadCap, + }, + [{ + area: { + includedSubspaceId: TestSubspace.Gemma, + pathPrefix: [new Uint8Array([1])], + timeRange: { + start: BigInt(0), + end: OPEN_END, + }, + }, + maxCount: 0, + maxSize: BigInt(0), + }], + ]]), + transformPayload: (bytes) => bytes, + processReceivedPayload: (bytes) => bytes, + }); + + await delay(100 * scenario.timeMultiplier); + + const range: Range3d = { + subspaceRange: { + start: TestSubspace.Gemma, + end: TestSubspace.Dalton, + }, + pathRange: { + start: [], + end: OPEN_END, + }, + timeRange: { + start: 0n, + end: OPEN_END, + }, + }; + + const { size: alfieFamilySize } = await storeFamilyAlfie + .summarise(range); + const { size: bettyFamilySize } = await storeFamilyBetty + .summarise(range); + + assertEquals(alfieFamilySize, 100); + assertEquals(bettyFamilySize, 100); + + let actualSizeFamilyAlfie = 0; + let actualSizeFamilyBetty = 0; + let alfieHasAllPayloads = true; + let bettyHasAllPayloads = true; + + for await ( + const [, payload] of storeFamilyAlfie.query({ + area: { + includedSubspaceId: TestSubspace.Gemma, + pathPrefix: [], + timeRange: { + start: 0n, + end: OPEN_END, + }, + }, + maxCount: 0, + maxSize: 0n, + }, "subspace") + ) { + actualSizeFamilyAlfie += 1; + + if (!payload) { + alfieHasAllPayloads = false; + } + } + + for await ( + const [, payload] of storeFamilyBetty.query({ + area: { + includedSubspaceId: TestSubspace.Gemma, + pathPrefix: [], + timeRange: { + start: 0n, + end: OPEN_END, + }, + }, + maxCount: 0, + maxSize: 0n, + }, "subspace") + ) { + actualSizeFamilyBetty += 1; + + if (!payload) { + bettyHasAllPayloads = false; + } + } + + assertEquals(actualSizeFamilyAlfie, 100); + assertEquals(actualSizeFamilyBetty, 100); + assert(alfieHasAllPayloads, "Alfie does not have all payloads"); + assert(bettyHasAllPayloads, "Betty does not have all payloads"); + + messengerAlfie.close(); + messengerBetty.close(); + }); + + await scenario.dispose(); + }, + ); + Deno.test( `Reconciliation of many namespaces (${scenario.name})`, async (test) => { diff --git a/src/wgps/wgps_messenger.ts b/src/wgps/wgps_messenger.ts index f086547..9a4d70a 100644 --- a/src/wgps/wgps_messenger.ts +++ b/src/wgps/wgps_messenger.ts @@ -4,6 +4,7 @@ import { WgpsMessageValidationError, WillowError, } from "../errors.ts"; +import { RangeCounter } from "./range_counter.ts"; import { decodeMessages } from "./decoding/decode_messages.ts"; import { MessageEncoder } from "./encoding/message_encoder.ts"; import { ReadyTransport } from "./ready_transport.ts"; @@ -311,7 +312,8 @@ export class WgpsMessenger< // Reconciliation - private yourRangeCounter = 0; + private yourRangeCounter = new RangeCounter(); + private myRangeCounter = new RangeCounter(); private getStore: GetStoreFn< Prefingerprint, @@ -1016,6 +1018,8 @@ export class WgpsMessenger< private setupReconciliation() { // When our announcer releases an 'announcement pack' (everything needed to announce and send some entries)... onAsyncIterate(this.announcer.announcementPacks(), async (pack) => { + this.yourRangeCounter.done(pack.announcement.covers); + // Bind any static tokens first. for (const staticToken of pack.staticTokenBinds) { this.encoder.encode({ @@ -1036,6 +1040,10 @@ export class WgpsMessenger< covers: pack.announcement.covers, }); + if (pack.announcement.wantResponse) { + this.myRangeCounter.getNext(); + } + // Then send the entries. for (const entry of pack.entries) { this.encoder.encode({ @@ -1100,6 +1108,8 @@ export class WgpsMessenger< ({ fingerprint, range, covers }) => { // Send a ReconciliationSendFingerprint message + this.yourRangeCounter.done(covers); + this.encoder.encode({ kind: MsgKind.ReconciliationSendFingerprint, fingerprint, @@ -1108,6 +1118,8 @@ export class WgpsMessenger< receiverHandle: intersection.theirs, covers, }); + + this.myRangeCounter.getNext(); }, ); @@ -1333,14 +1345,14 @@ export class WgpsMessenger< message.senderHandle, ); + this.myRangeCounter.done(message.covers); + await reconciler.respond( message.range, message.fingerprint, - this.yourRangeCounter, + Number(this.yourRangeCounter.getNext()), ); - this.yourRangeCounter += 1; - break; } case MsgKind.ReconciliationAnnounceEntries: { @@ -1364,6 +1376,8 @@ export class WgpsMessenger< range: message.range, }; + this.myRangeCounter.done(message.covers); + // If a response is wanted... queue up announcement if (message.wantResponse) { this.announcer.queueAnnounce({ @@ -1373,7 +1387,7 @@ export class WgpsMessenger< wantResponse: false, receiverHandle: message.senderHandle, senderHandle: message.receiverHandle, - covers: message.covers, + covers: this.yourRangeCounter.getNext(), }); } From 312ad3429041e9245837c29ea3e9a99f7b247c26 Mon Sep 17 00:00:00 2001 From: xash Date: Thu, 21 Nov 2024 08:00:25 +0100 Subject: [PATCH 2/3] Add WGPS reconciliation tracking via events --- src/wgps/events.ts | 11 +++++++++++ src/wgps/wgps_messenger.ts | 19 ++++++++++++++++++- 2 files changed, 29 insertions(+), 1 deletion(-) create mode 100644 src/wgps/events.ts diff --git a/src/wgps/events.ts b/src/wgps/events.ts new file mode 100644 index 0000000..3b108fc --- /dev/null +++ b/src/wgps/events.ts @@ -0,0 +1,11 @@ +export class WgpsStatusEvent + extends CustomEvent<{ remaining: number; all: number }> { + constructor(remaining: number, all: number) { + super("status", { + detail: { + remaining, + all, + }, + }); + } +} diff --git a/src/wgps/wgps_messenger.ts b/src/wgps/wgps_messenger.ts index 9a4d70a..336e6bc 100644 --- a/src/wgps/wgps_messenger.ts +++ b/src/wgps/wgps_messenger.ts @@ -30,6 +30,7 @@ import { } from "./types.ts"; import type { Intersection } from "./pai/types.ts"; import { onAsyncIterate } from "./util.ts"; +import { WgpsStatusEvent } from "./events.ts"; import { GuaranteedQueue } from "./guaranteed_queue.ts"; import { AoiIntersectionFinder } from "./reconciliation/aoi_intersection_finder.ts"; @@ -186,7 +187,7 @@ export class WgpsMessenger< SubspaceId, PayloadDigest, AuthorisationOpts, -> { +> extends EventTarget { private closed = false; private interests: Map< @@ -436,6 +437,8 @@ export class WgpsMessenger< AuthorisationOpts >, ) { + super(); + if (opts.maxPayloadSizePower < 0 || opts.maxPayloadSizePower > 64) { throw new ValidationError( "maxPayloadSizePower must be a natural number less than or equal to 64", @@ -1065,6 +1068,8 @@ export class WgpsMessenger< kind: MsgKind.ReconciliationTerminatePayload, }); } + + this.dispatchEvent(this.status()); }); // Whenever the area of interest intersection finder finds an intersection from the setup phase... @@ -1120,6 +1125,7 @@ export class WgpsMessenger< }); this.myRangeCounter.getNext(); + this.dispatchEvent(this.status()); }, ); @@ -1353,6 +1359,7 @@ export class WgpsMessenger< Number(this.yourRangeCounter.getNext()), ); + this.dispatchEvent(this.status()); break; } case MsgKind.ReconciliationAnnounceEntries: { @@ -1391,6 +1398,7 @@ export class WgpsMessenger< }); } + this.dispatchEvent(this.status()); break; } case MsgKind.ReconciliationSendEntry: { @@ -1684,6 +1692,15 @@ export class WgpsMessenger< this.handlesStaticTokensTheirs.bind(message.staticToken); } + status(): WgpsStatusEvent { + const yourInfo = this.yourRangeCounter.info(); + const myInfo = this.myRangeCounter.info(); + return new WgpsStatusEvent( + Number(yourInfo.remaining + myInfo.remaining), + Number(yourInfo.all + myInfo.all), + ); + } + close() { this.closed = true; this.transport.close(); From 8008e7820d46176c70a3a9f444036e66dd71d6a6 Mon Sep 17 00:00:00 2001 From: xash Date: Thu, 21 Nov 2024 08:27:18 +0100 Subject: [PATCH 3/3] Speed up WGPS tests by tracking reconciliation status Currently the status will indicate done even when some entries are yet to be processed, e.g. storing them. That's why there is still a small(er) delay. --- src/wgps/wgps_messenger.test.ts | 46 +++++++++++++++++++++++++++++---- 1 file changed, 41 insertions(+), 5 deletions(-) diff --git a/src/wgps/wgps_messenger.test.ts b/src/wgps/wgps_messenger.test.ts index d3cb3d1..cd1f84b 100644 --- a/src/wgps/wgps_messenger.test.ts +++ b/src/wgps/wgps_messenger.test.ts @@ -29,6 +29,7 @@ import { emptyDir, ensureDir } from "@std/fs"; import { ANY_SUBSPACE, OPEN_END, type Range3d } from "@earthstar/willow-utils"; import { assert, assertEquals, assertNotEquals } from "@std/assert"; import { encodeBase64 } from "@std/encoding/base64"; +import { WgpsStatusEvent } from "./events.ts"; type WgpsScenario = { name: string; @@ -1756,7 +1757,11 @@ function testWgpsMessenger(scenario: WgpsScenario) { processReceivedPayload: (bytes) => bytes, }); - await delay(20 * scenario.timeMultiplier); + await Promise.all([ + reconciliationDone(messengerAlfie), + reconciliationDone(messengerBetty), + ]); + await delay(10 * scenario.timeMultiplier); const range: Range3d = { subspaceRange: { @@ -2037,7 +2042,11 @@ function testWgpsMessenger(scenario: WgpsScenario) { processReceivedPayload: (bytes) => bytes, }); - await delay(20 * scenario.timeMultiplier); + await Promise.all([ + reconciliationDone(messengerAlfie), + reconciliationDone(messengerBetty), + ]); + await delay(10 * scenario.timeMultiplier); const range: Range3d = { subspaceRange: { @@ -2327,7 +2336,11 @@ function testWgpsMessenger(scenario: WgpsScenario) { processReceivedPayload: (bytes) => bytes, }); - await delay(20 * scenario.timeMultiplier); + await Promise.all([ + reconciliationDone(messengerAlfie), + reconciliationDone(messengerBetty), + ]); + await delay(10 * scenario.timeMultiplier); const range: Range3d = { subspaceRange: { @@ -2625,7 +2638,11 @@ function testWgpsMessenger(scenario: WgpsScenario) { }, }); - await delay(20 * scenario.timeMultiplier); + await Promise.all([ + reconciliationDone(messengerAlfie), + reconciliationDone(messengerBetty), + ]); + await delay(10 * scenario.timeMultiplier); const range: Range3d = { subspaceRange: { @@ -2844,7 +2861,11 @@ function testWgpsMessenger(scenario: WgpsScenario) { processReceivedPayload: (bytes) => bytes, }); - await delay(100 * scenario.timeMultiplier); + await Promise.all([ + reconciliationDone(messengerAlfie), + reconciliationDone(messengerBetty), + ]); + await delay(10 * scenario.timeMultiplier); const range: Range3d = { subspaceRange: { @@ -3456,3 +3477,18 @@ export class StoreMap< return newStore; } } + +function reconciliationDone(wgps: any) { + return new Promise((resolve) => { + const status = wgps.status(); + if (status.detail.remaining === 0 && status.detail.all > 0) { + resolve(true); + } else { + wgps.addEventListener("status", (e: WgpsStatusEvent) => { + if (e.detail.remaining === 0 && e.detail.all > 0) { + resolve(true); + } + }); + } + }); +}