Add some supporting stuff for future plum trees

src/store/events.ts

@@ -36,16 +36,19 @@ export class EntryIngestEvent<
   {
     entry: Entry<NamespacePublicKey, SubspacePublicKey, PayloadDigest>;
     authToken: AuthorisationToken;
+    externalSourceId?: string;
   }
 > {
   constructor(
     entry: Entry<NamespacePublicKey, SubspacePublicKey, PayloadDigest>,
     authToken: AuthorisationToken,
+    externalSourceId?: string,
   ) {
     super("entryingest", {
       detail: {
         entry,
         authToken,
+        externalSourceId,
       },
     });
   }
@@ -61,17 +64,43 @@ export class PayloadIngestEvent<
   entry: Entry<NamespacePublicKey, SubspacePublicKey, PayloadDigest>;
   authToken: AuthorisationToken;
   payload: Payload;
+  externalSourceId?: string;
 }> {
   constructor(
     entry: Entry<NamespacePublicKey, SubspacePublicKey, PayloadDigest>,
     authToken: AuthorisationToken,
     payload: Payload,
+    externalSourceId?: string,
   ) {
     super("payloadingest", {
       detail: {
         entry,
         authToken,
         payload,
+        externalSourceId,
+      },
+    });
+  }
+}
+
+/** Emitted after a {@linkcode Store} attempts to ingest a payload, but already had it. */
+export class PayloadNoOpEvent<
+  NamespacePublicKey,
+  SubspacePublicKey,
+  PayloadDigest,
+  AuthorisationToken,
+> extends CustomEvent<{
+  entry: Entry<NamespacePublicKey, SubspacePublicKey, PayloadDigest>;
+  externalSourceId?: string;
+}> {
+  constructor(
+    entry: Entry<NamespacePublicKey, SubspacePublicKey, PayloadDigest>,
+    externalSourceId?: string,
+  ) {
+    super("payloadnoop", {
+      detail: {
+        entry,
+        externalSourceId,
       },
     });
   }

src/store/store.ts

@@ -14,6 +14,7 @@ import {
   EntryPayloadSetEvent,
   EntryRemoveEvent,
   PayloadIngestEvent,
+  PayloadNoOpEvent,
   PayloadRemoveEvent,
 } from "./events.ts";
 import type { Storage3d } from "./storage/storage_3d/types.ts";
@@ -618,6 +619,7 @@ export class Store<
     payload: AsyncIterable<Uint8Array>,
     allowPartial = false,
     offset = 0,
+    externalSourceId? : string,
   ): Promise<IngestPayloadEvent> {
     const getResult = await this.storage.get(
       entryDetails.subspace,
@@ -636,6 +638,9 @@ export class Store<
     const existingPayload = await this.payloadDriver.get(entry.payloadDigest);
 
     if (existingPayload) {
+      this.dispatchEvent(
+        new PayloadNoOpEvent(entry, externalSourceId),
+      );
       return {
         kind: "no_op",
         reason: "already_have_it",
@@ -692,7 +697,7 @@ export class Store<
       }
 
       this.dispatchEvent(
-        new PayloadIngestEvent(entry, authToken, complete),
+        new PayloadIngestEvent(entry, authToken, complete, externalSourceId),
       );
     }
 

src/store/types.ts

@@ -179,6 +179,8 @@ export type IngestEventFailure = {
 export type IngestEventNoOp = {
   kind: "no_op";
   reason: "obsolete_from_same_subspace" | "newer_prefix_found";
+  /** An ID representing the source of this ingested entry. */
+  externalSourceId?: string;
 };
 
 /** Emitted after a successuful entry ingestion. */
@@ -261,11 +263,15 @@ export type IngestPayloadEventFailure = {
 export type IngestPayloadEventNoOp = {
   kind: "no_op";
   reason: "already_have_it";
+  /** An ID representing the source of this ingested entry. */
+  externalSourceId?: string;
 };
 
 /** Emitted after the succesful ingestion of a payload. */
 export type IngestPayloadEventSuccess = {
   kind: "success";
+  /** An ID representing the source of this ingested entry. */
+  externalSourceId?: string;
 };
 
 /** Emitted after payload entry. */

src/wgps/data/payload_ingester.ts

@@ -44,6 +44,8 @@ export class PayloadIngester<
     | Entry<NamespaceId, SubspaceId, PayloadDigest>
     | null = null;
 
+  private id: string;
+
   constructor(opts: {
     getStore: GetStoreFn<
       Prefingerprint,
@@ -58,9 +60,12 @@ export class PayloadIngester<
       bytes: Uint8Array,
       entryLength: bigint,
     ) => Uint8Array;
+    id: string;
   }) {
     this.processReceivedPayload = opts.processReceivedPayload;
 
+    this.id = opts.id;
+
     onAsyncIterate(this.events, async (event) => {
       if (event === CANCELLATION) {
         if (this.currentIngestion.kind === "active") {
@@ -105,7 +110,7 @@ export class PayloadIngester<
             path: entry.path,
             subspace: entry.subspaceId,
             timestamp: entry.timestamp,
-          }, new CancellableIngestion(fifo));
+          }, new CancellableIngestion(fifo), false, 0, this.id);
 
           this.currentIngestion = {
             kind: "active",

src/wgps/reconciliation/announcer.ts

@@ -1,11 +1,10 @@
 import type { Range3d } from "@earthstar/willow-utils";
-import { encodeBase64 } from "@std/encoding/base64";
 import { FIFO } from "@korkje/fifo";
 import { WillowError } from "../../errors.ts";
 import type { Store } from "../../store/store.ts";
 import type { LengthyEntry, Payload, PayloadScheme } from "../../store/types.ts";
-import type { HandleStore } from "../handle_store.ts";
 import type { AuthorisationTokenScheme, COVERS_NONE } from "../types.ts";
+import type { StaticTokenStore } from "../static_token_store.ts";
 
 export type AnnouncerOpts<
   AuthorisationToken,
@@ -19,7 +18,7 @@ export type AnnouncerOpts<
     DynamicToken
   >;
   payloadScheme: PayloadScheme<PayloadDigest>;
-  staticTokenHandleStoreOurs: HandleStore<StaticToken>;
+  staticTokenStore: StaticTokenStore<StaticToken>,
 };
 
 type AnnouncementPack<
@@ -29,9 +28,7 @@ type AnnouncementPack<
   SubspaceId,
   PayloadDigest,
 > = {
-  // Send these first in SetupBindStaticToken messages
-  staticTokenBinds: StaticToken[];
-  // Then send a ReconciliationAnnounceEntries
+  // Send ReconciliationAnnounceEntries
   announcement: {
     range: Range3d<SubspaceId>;
     count: number;
@@ -66,9 +63,7 @@ export class Announcer<
     DynamicToken
   >;
   private payloadScheme: PayloadScheme<PayloadDigest>;
-  private staticTokenHandleStoreOurs: HandleStore<StaticToken>;
-
-  private staticTokenHandleMap = new Map<string, bigint>();
+  private staticTokenStore: StaticTokenStore<StaticToken>;
 
   private announcementPackQueue = new FIFO<
     AnnouncementPack<
@@ -92,33 +87,7 @@ export class Announcer<
   ) {
     this.authorisationTokenScheme = opts.authorisationTokenScheme;
     this.payloadScheme = opts.payloadScheme;
-    this.staticTokenHandleStoreOurs = opts.staticTokenHandleStoreOurs;
-  }
-
-  private getStaticTokenHandle(
-    staticToken: StaticToken,
-  ): { handle: bigint; alreadyExisted: boolean } {
-    const encoded = this.authorisationTokenScheme.encodings.staticToken.encode(
-      staticToken,
-    );
-    const base64 = encodeBase64(encoded);
-
-    const existingHandle = this.staticTokenHandleMap.get(base64);
-
-    if (existingHandle !== undefined) {
-      const canUse = this.staticTokenHandleStoreOurs.canUse(existingHandle);
-
-      if (!canUse) {
-        throw new WillowError("Could not use a static token handle");
-      }
-
-      return { handle: existingHandle, alreadyExisted: true };
-    }
-
-    const newHandle = this.staticTokenHandleStoreOurs.bind(staticToken);
-    this.staticTokenHandleMap.set(base64, newHandle);
-
-    return { handle: newHandle, alreadyExisted: false };
+    this.staticTokenStore = opts.staticTokenStore;
   }
 
   async queueAnnounce(announcement: {
@@ -140,7 +109,6 @@ export class Announcer<
   }) {
     // Queue announcement message.
 
-    const staticTokenBinds: StaticToken[] = [];
     const entries: {
       lengthyEntry: LengthyEntry<NamespaceId, SubspaceId, PayloadDigest>;
       staticTokenHandle: bigint;
@@ -158,14 +126,7 @@ export class Announcer<
       const [staticToken, dynamicToken] = this.authorisationTokenScheme
         .decomposeAuthToken(authToken);
 
-      const {
-        handle: staticTokenHandle,
-        alreadyExisted: staticTokenHandleAlreadyExisted,
-      } = this.getStaticTokenHandle(staticToken);
-
-      if (!staticTokenHandleAlreadyExisted) {
-        staticTokenBinds.push(staticToken);
-      }
+      const staticTokenHandle = this.staticTokenStore.getByValue(staticToken);
 
       let available = payload ? await payload.length() : 0n;
 
@@ -187,7 +148,6 @@ export class Announcer<
     }
 
     this.announcementPackQueue.push({
-      staticTokenBinds,
       announcement: {
         count: entries.length,
         range: announcement.range,

src/wgps/static_token_store.ts

@@ -0,0 +1,49 @@
+import { encodeBase64 } from "@std/encoding/base64";
+import { HandleStore } from "./handle_store.ts";
+import { WillowError } from "../errors.ts";
+import FIFO from "@korkje/fifo";
+
+/** Maps static tokens to handles and vice versa. `boundStaticTokens` needs to be sent to the peer. */
+export class StaticTokenStore<StaticToken> {
+  private byHandle = new HandleStore<StaticToken>();
+  private byValue = new Map<string, bigint>();
+
+  private valueEncoder: (value: StaticToken) => Uint8Array;
+  private boundTokensQueue = new FIFO<StaticToken>();
+
+  constructor(valueEncoder: (value: StaticToken) => Uint8Array) {
+    this.valueEncoder = valueEncoder;
+  }
+
+  getByValue(
+    value: StaticToken,
+  ): bigint {
+    const encoded = this.valueEncoder(value);
+    const base64 = encodeBase64(encoded);
+
+    const existingHandle = this.byValue.get(base64);
+
+    if (existingHandle !== undefined) {
+      const canUse = this.byHandle.canUse(existingHandle);
+
+      if (!canUse) {
+        throw new WillowError("Could not use a static token handle");
+      }
+
+      return existingHandle;
+    }
+
+    const newHandle = this.byHandle.bind(value);
+    this.byValue.set(base64, newHandle);
+
+    this.boundTokensQueue.push(value);
+
+    return newHandle;
+  }
+
+  *boundStaticTokens() {
+    for (const boundToken of this.boundTokensQueue) {
+      yield boundToken;
+    }
+  }
+}