chore: different periods between send and receive

src/lib/controller/store.ts

@@ -12,7 +12,8 @@
 import { fillStoreCache, sendUpdatesAndFlushCache } from "./storeCache";
 
 const namespace = "pepr-system";
-export const debounceBackoff = 1000;
+const debounceBackoffReceive = 1000;
+const debounceBackoffSend = 4000;
 
 export class StoreController {
   #name: string;
@@ -68,6 +69,15 @@
 
   #migrateAndSetupWatch = async (store: Store): Promise<void> => {
     Log.debug(redactedStore(store), "Pepr Store migration");
+    // Add cacheID label to store
+    await K8s(Store, { namespace, name: this.#name }).Patch([
+      {
+        op: "add",
+        path: "/metadata/labels/pepr.dev-cacheID",
+        value: `${Date.now()}`,
+      },
+    ]);
+
     const data: DataStore = store.data || {};
     let storeCache: Record<string, Operation> = {};
 
@@ -134,7 +144,7 @@
 
     // Debounce the update to 1 second to avoid multiple rapid calls
     clearTimeout(this.#sendDebounce);
-    this.#sendDebounce = setTimeout(debounced, this.#onReady ? 0 : debounceBackoff);
+    this.#sendDebounce = setTimeout(debounced, this.#onReady ? 0 : debounceBackoffReceive);
   };
 
   #send = (capabilityName: string): DataSender => {
@@ -151,7 +161,7 @@
         Log.debug(redactedPatch(storeCache), "Sending updates to Pepr store");
         void sendUpdatesAndFlushCache(storeCache, namespace, this.#name);
       }
-    }, debounceBackoff);
+    }, debounceBackoffSend);
 
     return sender;
   };
@@ -165,6 +175,9 @@
         metadata: {
           name: this.#name,
           namespace,
+          labels: {
+            "pepr.dev-cacheID": `${Date.now()}`,
+          },
         },
         data: {
           // JSON Patch will die if the data is empty, so we need to add a placeholder

src/lib/controller/storeCache.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it, jest, afterEach } from "@jest/globals";
-import { fillStoreCache, sendUpdatesAndFlushCache } from "./storeCache";
+import { fillStoreCache, sendUpdatesAndFlushCache, updateCacheID } from "./storeCache";
 import { Operation } from "fast-json-patch";
 import { GenericClass, K8s, KubernetesObject } from "kubernetes-fluent-client";
 import { K8sInit } from "kubernetes-fluent-client/dist/fluent/types";
@@ -100,3 +100,20 @@ describe("sendCache", () => {
     });
   });
 });
+
+describe("updateCacheId", () => {
+  it("should update the metadata label of the cacheID in the payload array of patches", () => {
+    const patches: Operation[] = [
+      {
+        op: "add",
+        path: "/data/hello-pepr-v2-a",
+        value: "a",
+      },
+    ];
+
+    const updatedPatches = updateCacheID(patches);
+    expect(updatedPatches.length).toBe(2);
+    expect(updatedPatches[1].op).toBe("replace");
+    expect(updatedPatches[1].path).toBe("/metadata/labels/pepr.dev-cacheID");
+  });
+});

src/lib/controller/storeCache.ts

@@ -11,7 +11,7 @@ export const sendUpdatesAndFlushCache = async (cache: Record<string, Operation>,
 
   try {
     if (payload.length > 0) {
-      await K8s(Store, { namespace, name }).Patch(payload); // Send patch to cluster
+      await K8s(Store, { namespace, name }).Patch(updateCacheID(payload)); // Send patch to cluster
       Object.keys(cache).forEach(key => delete cache[key]);
     }
   } catch (err) {
@@ -61,3 +61,12 @@ export const fillStoreCache = (
   }
   return cache;
 };
+
+export function updateCacheID(payload: Operation[]): Operation[] {
+  payload.push({
+    op: "replace",
+    path: "/metadata/labels/pepr.dev-cacheID",
+    value: `${Date.now()}`,
+  });
+  return payload;
+}

src/lib/schedule.test.ts

@@ -20,21 +20,21 @@ export class MockStorage {
     this.storage[key] = value;
   }
 
-  setItemAndWait(key: string, value: string): Promise<void> {
+  setItemAndWait(key: string, value: string): Promise<string> {
     return new Promise(resolve => {
       this.storage[key] = value;
-      resolve();
+      resolve("ok");
     });
   }
 
   removeItem(key: string): void {
     delete this.storage[key];
   }
 
-  removeItemAndWait(key: string): Promise<void> {
+  removeItemAndWait(key: string): Promise<string> {
     return new Promise(resolve => {
       delete this.storage[key];
-      resolve();
+      resolve("ok");
     });
   }
 

src/lib/storage.ts

@@ -12,6 +12,11 @@
 const MAX_WAIT_TIME = 15000;
 const STORE_VERSION_PREFIX = "v2";
 
+interface WaitRecord {
+  timeout?: ReturnType<typeof setTimeout>;
+  unsubscribe?: () => void;
+}
+
 export function v2StoreKey(key: string): string {
   return `${STORE_VERSION_PREFIX}-${pointer.escape(key)}`;
 }
@@ -58,13 +63,13 @@
    * Sets the value of the pair identified by key to value, creating a new key/value pair if none existed for key previously.
    * Resolves when the key/value show up in the store.
    */
-  setItemAndWait(key: string, value: string): Promise<void>;
+  setItemAndWait(key: string, value: string): Promise<string>;
 
   /**
    * Remove the value of the key.
    * Resolves when the key does not show up in the store.
    */
-  removeItemAndWait(key: string): Promise<void>;
+  removeItemAndWait(key: string): Promise<string>;
 }
 
 /**
@@ -128,22 +133,24 @@
    * @param value - The value of the key
    * @returns
    */
-  setItemAndWait = (key: string, value: string): Promise<void> => {
+  setItemAndWait = (key: string, value: string): Promise<string> => {
     this.#dispatchUpdate("add", [v2StoreKey(key)], value);
+    const record: WaitRecord = {};
 
-    return new Promise<void>((resolve, reject) => {
-      const unsubscribe = this.subscribe(data => {
+    return new Promise<string>((resolve, reject) => {
+      // If promise has not resolved before MAX_WAIT_TIME reject
+      record.timeout = setTimeout(() => {
+        record.unsubscribe!();
+        return reject(`MAX_WAIT_TIME elapsed: Key ${key} not seen in ${MAX_WAIT_TIME / 1000}s`);
+      }, MAX_WAIT_TIME);
+
+      record.unsubscribe = this.subscribe(data => {
         if (data[`${v2UnescapedStoreKey(key)}`] === value) {
-          unsubscribe();
-          resolve();
+          record.unsubscribe!();
+          clearTimeout(record.timeout);
+          resolve("ok");
         }
       });
-
-      // If promise has not resolved before MAX_WAIT_TIME reject
-      setTimeout(() => {
-        unsubscribe();
-        return reject();
-      }, MAX_WAIT_TIME);
     });
   };
 
@@ -154,21 +161,23 @@
    * @param key - The key to add into the store
    * @returns
    */
-  removeItemAndWait = (key: string): Promise<void> => {
+  removeItemAndWait = (key: string): Promise<string> => {
     this.#dispatchUpdate("remove", [v2StoreKey(key)]);
-    return new Promise<void>((resolve, reject) => {
-      const unsubscribe = this.subscribe(data => {
+    const record: WaitRecord = {};
+    return new Promise<string>((resolve, reject) => {
+      // If promise has not resolved before MAX_WAIT_TIME reject
+      record.timeout = setTimeout(() => {
+        record.unsubscribe!();
+        return reject(`MAX_WAIT_TIME elapsed: Key ${key} still seen after ${MAX_WAIT_TIME / 1000}s`);
+      }, MAX_WAIT_TIME);
+
+      record.unsubscribe = this.subscribe(data => {
         if (!Object.hasOwn(data, `${v2UnescapedStoreKey(key)}`)) {
-          unsubscribe();
-          resolve();
+          record.unsubscribe!();
+          clearTimeout(record.timeout);
+          resolve("ok");
         }
       });
-
-      // If promise has not resolved before MAX_WAIT_TIME reject
-      setTimeout(() => {
-        unsubscribe();
-        return reject();
-      }, MAX_WAIT_TIME);
     });
   };