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/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..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: { @@ -2653,6 +2670,286 @@ 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 Promise.all([ + reconciliationDone(messengerAlfie), + reconciliationDone(messengerBetty), + ]); + await delay(10 * 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) => { @@ -3180,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); + } + }); + } + }); +} diff --git a/src/wgps/wgps_messenger.ts b/src/wgps/wgps_messenger.ts index f086547..336e6bc 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"; @@ -29,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"; @@ -185,7 +187,7 @@ export class WgpsMessenger< SubspaceId, PayloadDigest, AuthorisationOpts, -> { +> extends EventTarget { private closed = false; private interests: Map< @@ -311,7 +313,8 @@ export class WgpsMessenger< // Reconciliation - private yourRangeCounter = 0; + private yourRangeCounter = new RangeCounter(); + private myRangeCounter = new RangeCounter(); private getStore: GetStoreFn< Prefingerprint, @@ -434,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", @@ -1016,6 +1021,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 +1043,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({ @@ -1057,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... @@ -1100,6 +1113,8 @@ export class WgpsMessenger< ({ fingerprint, range, covers }) => { // Send a ReconciliationSendFingerprint message + this.yourRangeCounter.done(covers); + this.encoder.encode({ kind: MsgKind.ReconciliationSendFingerprint, fingerprint, @@ -1108,6 +1123,9 @@ export class WgpsMessenger< receiverHandle: intersection.theirs, covers, }); + + this.myRangeCounter.getNext(); + this.dispatchEvent(this.status()); }, ); @@ -1333,14 +1351,15 @@ 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; - + this.dispatchEvent(this.status()); break; } case MsgKind.ReconciliationAnnounceEntries: { @@ -1364,6 +1383,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,10 +1394,11 @@ export class WgpsMessenger< wantResponse: false, receiverHandle: message.senderHandle, senderHandle: message.receiverHandle, - covers: message.covers, + covers: this.yourRangeCounter.getNext(), }); } + this.dispatchEvent(this.status()); break; } case MsgKind.ReconciliationSendEntry: { @@ -1670,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();