refactor(node-sdk): throttle flags cache refreshes

roncohen · roncohen · commit 29c556786c27 · 2026-04-06T21:46:26.000+02:00
diff --git a/packages/node-sdk/README.md b/packages/node-sdk/README.md
@@ -461,6 +461,8 @@ Reflag maintains a cached set of flag definitions in the memory of your worker w
 
 The SDK caches flag definitions in memory for fast performance. The first request to a new worker instance fetches definitions from Reflag's servers, while subsequent requests use the cache. When the cache expires, it's updated in the background. `ctx.waitUntil(reflag.flush())` ensures completion of the background work, so response times are not affected. This background work may increase wall-clock time for your worker, but it will not measurably increase billable CPU time on platforms like Cloudflare.
 
+`EdgeClient` uses `flagsSyncMode: "in-request"`. Refresh fetch starts are throttled to at most once per second, and Cloudflare Workers cannot rely on delayed timer callbacks to run follow-up refreshes later. That means `refreshFlags()` calls made during the throttle window only mark a refresh as pending, so the call itself may resolve before the fetch runs. The queued refresh runs on the next request/access or `refreshFlags()` call after the throttle window expires.
+
 ## Error Handling
 
 The SDK is designed to fail gracefully and never throw exceptions to the caller. Instead, it logs errors and provides
diff --git a/packages/node-sdk/src/client.ts b/packages/node-sdk/src/client.ts
@@ -503,35 +503,54 @@ export class ReflagClient {
       this._config.apiBaseUrl += "/";
     }
 
-    this.flagsCache = new FlagsCache(async (waitForVersion?: number) => {
-      const path =
-        waitForVersion === undefined
-          ? "features"
-          : `features?waitForVersion=${encodeURIComponent(String(waitForVersion))}`;
-
-      const res = await this.get<FlagsAPIResponse>(
-        path,
-        this._config.flagsFetchRetries,
-      );
-      if (
-        !isObject(res) ||
-        !Array.isArray(res?.features) ||
-        !Number.isInteger(res?.flagStateVersion) ||
-        res.flagStateVersion < 0
-      ) {
-        const fallbackDefinitions = await this.loadFlagsFallbackDefinitions();
-        return fallbackDefinitions
-          ? { definitions: fallbackDefinitions }
-          : undefined;
-      }
+    this.flagsCache = new FlagsCache(
+      async (waitForVersion?: number) => {
+        const path =
+          waitForVersion === undefined
+            ? "features"
+            : `features?waitForVersion=${encodeURIComponent(String(waitForVersion))}`;
+
+        const res = await this.get<FlagsAPIResponse>(
+          path,
+          this._config.flagsFetchRetries,
+        );
+        if (
+          !isObject(res) ||
+          !Array.isArray(res?.features) ||
+          !Number.isInteger(res?.flagStateVersion) ||
+          res.flagStateVersion < 0
+        ) {
+          const fallbackDefinitions = await this.loadFlagsFallbackDefinitions();
+          return fallbackDefinitions
+            ? { definitions: fallbackDefinitions }
+            : undefined;
+        }
 
-      void this.saveFlagsFallbackDefinitions(res.features);
-      this.canLoadFlagsFallbackProvider = false;
-      return {
-        definitions: compileFlagDefinitions(res.features),
-        flagStateVersion: res.flagStateVersion,
-      };
-    }, this.logger);
+        void this.saveFlagsFallbackDefinitions(res.features);
+        this.canLoadFlagsFallbackProvider = false;
+        return {
+          definitions: compileFlagDefinitions(res.features),
+          flagStateVersion: res.flagStateVersion,
+        };
+      },
+      {
+        logger: this.logger,
+        // Edge runtimes use `in-request` mode and cannot rely on delayed timer
+        // callbacks to perform trailing refresh work after the current request.
+        scheduleTrailingRefresh:
+          flagsSyncMode === "in-request"
+            ? undefined
+            : (delayMs, callback) => {
+                const timer = setTimeout(callback, delayMs);
+                timer.unref?.();
+                return {
+                  cancel: () => {
+                    clearTimeout(timer);
+                  },
+                };
+              },
+      },
+    );
 
     this.flagsSyncController = createFlagsSyncController({
       mode: this._config.flagsSyncMode,
@@ -903,7 +922,18 @@ export class ReflagClient {
    *
    * Note: updated flag rules take a few seconds to propagate to all servers.
    *
-   * Concurrent calls are deduplicated — multiple calls share the same in-flight request.
+   * Fetch starts are throttled to at most once per second.
+   * Outside `in-request` mode, throttling only delays when the next fetch
+   * begins: `refreshFlags(99)` still waits until the cache has applied flag
+   * definitions from version `99` or newer before the promise resolves.
+   * Concurrent callers are deduplicated and may share the same in-flight or
+   * scheduled follow-up refresh.
+   *
+   * In `in-request` mode, delayed follow-up refreshes are not scheduled. On edge
+   * runtimes like Cloudflare Workers, a call during the throttle window only
+   * records pending refresh work, and the promise may resolve before that fetch
+   * runs. The pending refresh is executed on the next request/access or
+   * `refreshFlags()` call after the throttle window expires.
    *
    * @param waitForVersion - Optional flag state version to wait for before returning updated definitions.
    */
@@ -1817,6 +1847,14 @@ export class BoundReflagClient {
   /**
    * Refreshes the flag definitions from the server.
    *
+   * Fetch starts are throttled to at most once per second. Outside
+   * `in-request` mode, a call to `refreshFlags(waitForVersion)` still waits
+   * until the cache has applied that version or newer.
+   *
+   * In `in-request` mode, a throttled call only records pending refresh work.
+   * The fetch then runs on the next request/access or `refreshFlags()` call
+   * after the throttle window expires.
+   *
    * @param waitForVersion - Optional flag state version to wait for before returning updated definitions.
    */
   public async refreshFlags(waitForVersion?: number) {
diff --git a/packages/node-sdk/src/edgeClient.ts b/packages/node-sdk/src/edgeClient.ts
@@ -10,6 +10,12 @@ export type EdgeClientOptions = Omit<
  * The EdgeClient is ReflagClient pre-configured to be used in edge runtimes, like
  * Cloudflare Workers.
  *
+ * It always uses `flagsSyncMode: "in-request"`. Refresh fetch starts are
+ * throttled to at most once per second. A `refreshFlags()` call made during
+ * that throttle window only records pending refresh work, so the call may
+ * resolve before the fetch runs. That pending refresh is executed on the next
+ * request/access or `refreshFlags()` call after the window expires.
+ *
  * @example
  * ```ts
  * // set the REFLAG_SECRET_KEY environment variable or pass the secret key in the constructor
diff --git a/packages/node-sdk/src/flags-cache.ts b/packages/node-sdk/src/flags-cache.ts
@@ -1,34 +1,65 @@
 import type { CachedFlagDefinition, Logger } from "./types";
 
+const DEFAULT_MIN_REFRESH_INTERVAL_MS = 1000;
+
 type FlagsCacheRefreshResult = {
   definitions: CachedFlagDefinition[];
   flagStateVersion?: number;
 };
 
+type FlagsCacheScheduledRefresh = {
+  cancel: () => void;
+};
+
+type FlagsCacheOptions = {
+  logger?: Logger;
+  minRefreshIntervalMs?: number;
+  scheduleTrailingRefresh?: (
+    delayMs: number,
+    callback: () => void,
+  ) => FlagsCacheScheduledRefresh;
+};
+
 /**
  * Stores the latest compiled flag definitions and coordinates refresh work.
  *
- * A single instance is shared across all sync modes. We allow at most one
- * in-flight fetch plus one pending follow-up refresh. Response
+ * A single instance is shared across all sync modes. Response
  * `flagStateVersion`s decide whether fetched definitions replace the current
  * cache, so correctness does not depend on request ordering.
+ *
+ * Refreshes are throttled to at most one fetch start per interval. When the
+ * runtime supports delayed work we schedule one trailing refresh; otherwise we
+ * keep the pending refresh queued until the next caller touches the cache.
  */
 export class FlagsCache {
   private value: CachedFlagDefinition[] | undefined;
   private flagStateVersion: number | undefined;
   private refreshPromise: Promise<void> | undefined;
+  private scheduledRefresh: FlagsCacheScheduledRefresh | undefined;
+  private scheduledRefreshPromise: Promise<void> | undefined;
+  private resolveScheduledRefreshPromise: (() => void) | undefined;
   private lastRefreshAt: number | undefined;
+  private lastRefreshStartedAt: number | undefined;
   private destroyed = false;
 
   private pendingFullRefresh = false;
   private pendingWaitForVersion: number | undefined;
 
+  private readonly logger?: Logger;
+  private readonly minRefreshIntervalMs: number;
+  private readonly scheduleTrailingRefresh?: FlagsCacheOptions["scheduleTrailingRefresh"];
+
   constructor(
     private readonly fetchFlags: (
       waitForVersion?: number,
     ) => Promise<FlagsCacheRefreshResult | undefined>,
-    private readonly logger?: Logger,
-  ) {}
+    options: FlagsCacheOptions = {},
+  ) {
+    this.logger = options.logger;
+    this.minRefreshIntervalMs =
+      options.minRefreshIntervalMs ?? DEFAULT_MIN_REFRESH_INTERVAL_MS;
+    this.scheduleTrailingRefresh = options.scheduleTrailingRefresh;
+  }
 
   public get() {
     return this.value;
@@ -40,30 +71,34 @@ export class FlagsCache {
     }
 
     this.queueRefresh(waitForVersion);
-    return await this.ensureRefreshRunning();
+    this.ensureRefreshStartedOrScheduled();
+    await this.waitForQueuedWork();
+    return this.value;
   }
 
   public async waitRefresh() {
-    await this.refreshPromise;
+    await this.waitForQueuedWork();
   }
 
   public destroy() {
     this.destroyed = true;
     this.value = undefined;
     this.flagStateVersion = undefined;
     this.refreshPromise = undefined;
+    this.cancelScheduledRefresh();
     this.pendingFullRefresh = false;
     this.pendingWaitForVersion = undefined;
     this.lastRefreshAt = undefined;
+    this.lastRefreshStartedAt = undefined;
   }
 
   public getLastRefreshAt() {
     return this.lastRefreshAt;
   }
 
-  // Remember the newest refresh request we still need to run after the current
-  // fetch finishes. Versioned refreshes win over plain refreshes because they
-  // carry a concrete target we can wait for.
+  // Remember the newest refresh request we still need to run. Versioned
+  // refreshes win over plain refreshes because they carry a concrete target we
+  // can wait for.
   private queueRefresh(waitForVersion?: number) {
     if (waitForVersion !== undefined) {
       if (
@@ -88,6 +123,10 @@ export class FlagsCache {
     this.pendingFullRefresh = true;
   }
 
+  private hasPendingRefresh() {
+    return this.pendingWaitForVersion !== undefined || this.pendingFullRefresh;
+  }
+
   private takeNextRefreshRequest() {
     if (this.pendingWaitForVersion !== undefined) {
       const waitForVersion = this.pendingWaitForVersion;
@@ -129,39 +168,103 @@ export class FlagsCache {
     }
   }
 
-  private async runQueuedRefreshes() {
-    while (!this.destroyed) {
-      const request = this.takeNextRefreshRequest();
-      if (!request) {
-        return;
-      }
+  private getNextRefreshDelayMs() {
+    if (
+      this.lastRefreshStartedAt === undefined ||
+      this.minRefreshIntervalMs <= 0
+    ) {
+      return 0;
+    }
 
-      const result = await this.fetchFlags(request.waitForVersion);
-      if (this.destroyed || !result) {
-        continue;
-      }
+    return Math.max(
+      0,
+      this.lastRefreshStartedAt + this.minRefreshIntervalMs - Date.now(),
+    );
+  }
+
+  private settleScheduledRefresh() {
+    this.scheduledRefresh = undefined;
+    this.scheduledRefreshPromise = undefined;
+    this.resolveScheduledRefreshPromise?.();
+    this.resolveScheduledRefreshPromise = undefined;
+  }
 
-      this.clearSatisfiedPendingVersion(result.flagStateVersion);
+  private cancelScheduledRefresh() {
+    this.scheduledRefresh?.cancel();
+    this.settleScheduledRefresh();
+  }
 
-      if (!this.shouldApplyRefreshResult(result.flagStateVersion)) {
-        continue;
-      }
+  private ensureScheduledRefresh(delayMs: number) {
+    if (!this.scheduleTrailingRefresh || this.scheduledRefresh) {
+      return;
+    }
+
+    this.scheduledRefreshPromise = new Promise<void>((resolve) => {
+      this.resolveScheduledRefreshPromise = resolve;
+    });
 
-      this.value = result.definitions;
-      this.flagStateVersion = result.flagStateVersion;
-      this.lastRefreshAt = Date.now();
-      this.logger?.info("refreshed flag definitions");
+    this.scheduledRefresh = this.scheduleTrailingRefresh(delayMs, () => {
+      this.settleScheduledRefresh();
+      this.ensureRefreshStartedOrScheduled();
+    });
+  }
+
+  private ensureRefreshStartedOrScheduled() {
+    if (this.destroyed || this.refreshPromise || !this.hasPendingRefresh()) {
+      return;
     }
+
+    const delayMs = this.getNextRefreshDelayMs();
+    if (delayMs > 0) {
+      this.ensureScheduledRefresh(delayMs);
+      return;
+    }
+
+    this.cancelScheduledRefresh();
+    this.startNextRefresh();
   }
 
-  private async ensureRefreshRunning() {
-    if (!this.refreshPromise) {
-      this.refreshPromise = this.runQueuedRefreshes().finally(() => {
-        this.refreshPromise = undefined;
-      });
+  private startNextRefresh() {
+    const request = this.takeNextRefreshRequest();
+    if (!request) {
+      return;
     }
 
-    await this.refreshPromise;
-    return this.value;
+    this.lastRefreshStartedAt = Date.now();
+    this.refreshPromise = this.fetchAndApplyRefresh(
+      request.waitForVersion,
+    ).finally(() => {
+      this.refreshPromise = undefined;
+      this.ensureRefreshStartedOrScheduled();
+    });
+  }
+
+  private async fetchAndApplyRefresh(waitForVersion?: number) {
+    const result = await this.fetchFlags(waitForVersion);
+    if (this.destroyed || !result) {
+      return;
+    }
+
+    this.clearSatisfiedPendingVersion(result.flagStateVersion);
+
+    if (!this.shouldApplyRefreshResult(result.flagStateVersion)) {
+      return;
+    }
+
+    this.value = result.definitions;
+    this.flagStateVersion = result.flagStateVersion;
+    this.lastRefreshAt = Date.now();
+    this.logger?.info("refreshed flag definitions");
+  }
+
+  private async waitForQueuedWork() {
+    while (!this.destroyed) {
+      const workPromise = this.refreshPromise ?? this.scheduledRefreshPromise;
+      if (!workPromise) {
+        return;
+      }
+
+      await workPromise;
+    }
   }
 }
diff --git a/packages/node-sdk/test/flags-cache.test.ts b/packages/node-sdk/test/flags-cache.test.ts
