regional replication

Author: conico974

examples/e2e/app-router/open-next.config.ts

@@ -5,13 +5,14 @@ import doQueue from "@opennextjs/cloudflare/overrides/queue/do-queue";
 
 export default defineCloudflareConfig({
   incrementalCache: kvIncrementalCache,
-  // With such a configuration, we could have up to 12 * (8 + 2) = 120 Durable Objects instances
+  // With such a configuration, we could have up to 6(all regions) * (2 * 2 * (8 + 2)) = 240 Durable Objects instances
   tagCache: shardedTagCache({
-    baseShardSize: 12,
+    baseShardSize: 2,
     enableShardReplication: true,
     shardReplicationOptions: {
       numberOfSoftReplicas: 8,
       numberOfHardReplicas: 2,
+      enableRegionalReplication: true,
     },
   }),
   queue: doQueue,

packages/cloudflare/src/api/overrides/tag-cache/do-sharded-tag-cache.spec.ts

@@ -1,6 +1,7 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 
 import doShardedTagCache, {
+  AVAILABLE_REGIONS,
   DEFAULT_HARD_REPLICAS,
   DEFAULT_SOFT_REPLICAS,
   TagCacheDOId,
@@ -13,6 +14,8 @@ const getMock = vi
   .fn()
   .mockReturnValue({ hasBeenRevalidated: hasBeenRevalidatedMock, writeTags: writeTagsMock });
 const waitUntilMock = vi.fn().mockImplementation(async (fn) => fn());
+// @ts-expect-error - We define it here only for the test
+globalThis.continent = undefined;
 const sendDLQMock = vi.fn();
 vi.mock("../../cloudflare-context", () => ({
   getCloudflareContext: () => ({
@@ -23,6 +26,10 @@ vi.mock("../../cloudflare-context", () => ({
       },
     },
     ctx: { waitUntil: waitUntilMock },
+    cf: {
+      // @ts-expect-error - We define it here only for the test
+      continent: globalThis.continent,
+    },
   }),
 }));
 
@@ -106,6 +113,84 @@ describe("DOShardedTagCache", () => {
         expect(secondDOId?.replicaId).toBeGreaterThanOrEqual(1);
         expect(secondDOId?.replicaId).toBeLessThanOrEqual(DEFAULT_HARD_REPLICAS);
       });
+
+      it("should generate one doIds, but in the default region", () => {
+        const cache = doShardedTagCache({
+          baseShardSize: 4,
+          enableShardReplication: true,
+          shardReplicationOptions: {
+            numberOfSoftReplicas: 2,
+            numberOfHardReplicas: 2,
+            enableRegionalReplication: true,
+          },
+        });
+        const shardedTagCollection = cache.groupTagsByDO({
+          tags: ["tag1", "_N_T_/tag1"],
+          generateAllReplicas: false,
+        });
+        expect(shardedTagCollection.length).toBe(2);
+        const firstDOId = shardedTagCollection[0]?.doId;
+        const secondDOId = shardedTagCollection[1]?.doId;
+
+        expect(firstDOId?.shardId).toBe("tag-soft;shard-3");
+        expect(firstDOId?.region).toBe("enam");
+        expect(secondDOId?.shardId).toBe("tag-hard;shard-1");
+        expect(secondDOId?.region).toBe("enam");
+
+        // We still need to check if the last part is between the correct boundaries
+        expect(firstDOId?.replicaId).toBeGreaterThanOrEqual(1);
+        expect(firstDOId?.replicaId).toBeLessThanOrEqual(DEFAULT_SOFT_REPLICAS);
+
+        expect(secondDOId?.replicaId).toBeGreaterThanOrEqual(1);
+        expect(secondDOId?.replicaId).toBeLessThanOrEqual(DEFAULT_HARD_REPLICAS);
+      });
+
+      it("should generate one doIds, but in the correct region", () => {
+        // @ts-expect-error - We define it here only for the test
+        globalThis.continent = "EU";
+        const cache = doShardedTagCache({
+          baseShardSize: 4,
+          enableShardReplication: true,
+          shardReplicationOptions: {
+            numberOfSoftReplicas: 2,
+            numberOfHardReplicas: 2,
+            enableRegionalReplication: true,
+          },
+        });
+        const shardedTagCollection = cache.groupTagsByDO({
+          tags: ["tag1", "_N_T_/tag1"],
+          generateAllReplicas: false,
+        });
+        expect(shardedTagCollection.length).toBe(2);
+        expect(shardedTagCollection[0]?.doId.region).toBe("weur");
+        expect(shardedTagCollection[1]?.doId.region).toBe("weur");
+
+        //@ts-expect-error - We need to reset the global variable
+        globalThis.continent = undefined;
+      });
+
+      it("should generate all the appropriate replicas in all the regions with enableRegionalReplication", () => {
+        const cache = doShardedTagCache({
+          baseShardSize: 4,
+          enableShardReplication: true,
+          shardReplicationOptions: {
+            numberOfSoftReplicas: 2,
+            numberOfHardReplicas: 2,
+            enableRegionalReplication: true,
+          },
+        });
+        const shardedTagCollection = cache.groupTagsByDO({
+          tags: ["tag1", "_N_T_/tag1"],
+          generateAllReplicas: true,
+        });
+        // 6 regions times 4 shards replica
+        expect(shardedTagCollection.length).toBe(24);
+        shardedTagCollection.forEach(({ doId }) => {
+          expect(AVAILABLE_REGIONS).toContain(doId.region);
+          // It should end with the region
+          expect(doId.key).toMatch(/tag-(soft|hard);shard-\d;replica-\d;region-(enam|weur|sam|afr|apac|oc)$/);
+        });
+      });
     });
   });
 

packages/cloudflare/src/api/overrides/tag-cache/do-sharded-tag-cache.ts

@@ -11,6 +11,9 @@ export const DEFAULT_SOFT_REPLICAS = 4;
 export const DEFAULT_HARD_REPLICAS = 2;
 export const DEFAULT_WRITE_RETRIES = 3;
 export const DEFAULT_NUM_SHARDS = 4;
+export const DEFAULT_REGION = "enam" as const;
+export const AVAILABLE_REGIONS = ["enam", "weur", "apac", "sam", "afr", "oc"] as const;
+type AllowedDurableObjectRegion = (typeof AVAILABLE_REGIONS)[number];
 
 interface ShardedDOTagCacheOptions {
   /**
@@ -55,6 +58,21 @@ interface ShardedDOTagCacheOptions {
   shardReplicationOptions?: {
     numberOfSoftReplicas: number;
     numberOfHardReplicas: number;
+
+    /**
+     * Whether to enable regional replication
+     * Regional replication will duplicate each shards and their associated replicas into every regions
+     * This will reduce the latency for the read operations
+     * On write, the write will be sent to all the shards and all the replicas in all the regions
+     * @default false
+     */
+    enableRegionalReplication?: boolean;
+
+    /**
+     * Default region to use for the regional replication when the region cannot be determined
+     * @default "enam"
+     */
+    defaultRegion?: AllowedDurableObjectRegion;
   };
 
   /**
@@ -69,22 +87,25 @@ interface TagCacheDOIdOptions {
   numberOfReplicas: number;
   shardType: "soft" | "hard";
   replicaId?: number;
+  region?: DurableObjectLocationHint;
 }
 export class TagCacheDOId {
   shardId: string;
   replicaId: number;
+  region?: DurableObjectLocationHint;
   constructor(public options: TagCacheDOIdOptions) {
-    const { baseShardId, shardType, numberOfReplicas, replicaId } = options;
+    const { baseShardId, shardType, numberOfReplicas, replicaId, region } = options;
     this.shardId = `tag-${shardType};${baseShardId}`;
     this.replicaId = replicaId ?? this.generateRandomNumberBetween(1, numberOfReplicas);
+    this.region = region;
   }
 
   private generateRandomNumberBetween(min: number, max: number) {
     return Math.floor(Math.random() * (max - min + 1) + min);
   }
 
   get key() {
-    return `${this.shardId};replica-${this.replicaId}`;
+    return `${this.shardId};replica-${this.replicaId}${this.region ? `;region-${this.region}` : ""}`;
   }
 }
 class ShardedDOTagCache implements NextModeTagCache {
@@ -93,20 +114,28 @@ class ShardedDOTagCache implements NextModeTagCache {
   readonly numSoftReplicas: number;
   readonly numHardReplicas: number;
   readonly maxWriteRetries: number;
+  readonly enableRegionalReplication: boolean;
+  readonly defaultRegion: AllowedDurableObjectRegion;
   localCache?: Cache;
 
   constructor(private opts: ShardedDOTagCacheOptions = { baseShardSize: DEFAULT_NUM_SHARDS }) {
     this.numSoftReplicas = opts.shardReplicationOptions?.numberOfSoftReplicas ?? DEFAULT_SOFT_REPLICAS;
     this.numHardReplicas = opts.shardReplicationOptions?.numberOfHardReplicas ?? DEFAULT_HARD_REPLICAS;
     this.maxWriteRetries = opts.maxWriteRetries ?? DEFAULT_WRITE_RETRIES;
+    this.enableRegionalReplication = opts.shardReplicationOptions?.enableRegionalReplication ?? false;
+    this.defaultRegion = opts.shardReplicationOptions?.defaultRegion ?? DEFAULT_REGION;
   }
 
   private getDurableObjectStub(doId: TagCacheDOId) {
     const durableObject = getCloudflareContext().env.NEXT_TAG_CACHE_DO_SHARDED;
     if (!durableObject) throw new IgnorableError("No durable object binding for cache revalidation");
 
     const id = durableObject.idFromName(doId.key);
-    return durableObject.get(id);
+    debug("[shardedTagCache] - Accessing Durable Object : ", {
+      key: doId.key,
+      region: doId.region,
+    });
+    return durableObject.get(id, { locationHint: doId.region });
   }
 
   /**
@@ -134,7 +163,7 @@ class ShardedDOTagCache implements NextModeTagCache {
         ? Array.from({ length: numReplicas }, (_, i) => i + 1)
         : [undefined];
     }
-    return replicaIndexes.flatMap((replicaId) => {
+    const regionalReplicas = replicaIndexes.flatMap((replicaId) => {
       return tags
         .filter((tag) => (isSoft ? tag.startsWith(SOFT_TAG_PREFIX) : !tag.startsWith(SOFT_TAG_PREFIX)))
         .map((tag) => {
@@ -149,6 +178,51 @@ class ShardedDOTagCache implements NextModeTagCache {
           };
         });
     });
+    if (!this.enableRegionalReplication) return regionalReplicas;
+
+    // If we have regional replication enabled, we need to further duplicate the shards in all the regions
+    const regionalReplicasInAllRegions = generateAllReplicas
+      ? regionalReplicas.flatMap(({ doId, tag }) => {
+          return AVAILABLE_REGIONS.map((region) => {
+            return {
+              doId: new TagCacheDOId({
+                baseShardId: doId.options.baseShardId,
+                numberOfReplicas: numReplicas,
+                shardType,
+                replicaId: doId.replicaId,
+                region,
+              }),
+              tag,
+            };
+          });
+        })
+      : regionalReplicas.map(({ doId, tag }) => {
+          doId.region = this.getClosestRegion();
+          return { doId, tag };
+        });
+    return regionalReplicasInAllRegions;
+  }
+
+  getClosestRegion() {
+    const continent = getCloudflareContext().cf?.continent;
+    if (!continent) return this.defaultRegion;
+    debug("[shardedTagCache] - Continent : ", continent);
+    switch (continent) {
+      case "AF":
+        return "afr";
+      case "AS":
+        return "apac";
+      case "EU":
+        return "weur";
+      case "NA":
+        return "enam";
+      case "OC":
+        return "oc";
+      case "SA":
+        return "sam";
+      default:
+        return this.defaultRegion;
+    }
   }
 
   /**