Cache: port SWR tag revalidation from AWS #1122 / #1142

.changeset/swr-tag-revalidation.md

@@ -0,0 +1,17 @@
+---
+"@opennextjs/core": minor
+"@opennextjs/aws": minor
+---
+
+Port stale-while-revalidate tag revalidation from AWS
+
+Ports [#1122](https://github.com/opennextjs/opennextjs-aws/pull/1122) and
+[#1142](https://github.com/opennextjs/opennextjs-aws/pull/1142).
+
+Tags can now carry `stale` and `expire` durations, so an entry can be served stale while it
+revalidates instead of being dropped outright. `writeTags` accepts either a tag name or a
+`{ tag, stale, expire }` object, both tag cache flavours gained an optional `isStale`, and the
+composable cache handler implements the `updateTags` method added in Next.js 16.
+
+Tag cache overrides that need to deduplicate work within a request can use the new per-request
+`requestCache` on the OpenNext request context.

examples-cloudflare/e2e/experimental/e2e/use-cache.test.ts

@@ -26,21 +26,17 @@ test.describe("Composable Cache", () => {
 		expect(fullyCachedText).toEqual(initialFullyCachedText);
 	});
 
-	test("revalidateTag should work for fullyCached component", async ({ page, request }) => {
+	test("revalidateTag with SWR should reject unsupported tag caches", async ({ page, request }) => {
 		await page.goto("/use-cache/ssr");
 		const fullyCachedElt = page.getByTestId("fully-cached-with-tag");
 		await expect(fullyCachedElt).toBeVisible();
-
 		const initialFullyCachedText = await fullyCachedElt.textContent();
 
 		const resp = await request.get("/api/revalidate");
 		expect(resp.status()).toEqual(200);
-		expect(await resp.text()).toEqual("DONE");
 
 		await page.reload();
-		await expect(fullyCachedElt).toBeVisible();
-		const newFullyCachedText = await fullyCachedElt.textContent();
-		expect(newFullyCachedText).not.toEqual(initialFullyCachedText);
+		await expect(fullyCachedElt).toHaveText(initialFullyCachedText ?? "");
 	});
 
 	//TODO: figure out why it doesn't work in ISR (my guess is on our patch not working anymore in 16.1, but need investigation)

packages/aws/src/overrides/tagCache/dynamodb-lite.ts

@@ -9,13 +9,38 @@ import { AwsClient } from "aws4fetch";
 import { customFetchClient } from "../../utils/fetch.js";
 
 import { MAX_DYNAMO_BATCH_WRITE_ITEM_COUNT, getDynamoBatchWriteCommandConcurrency } from "./constants.js";
+import { isStaleRecord } from "./stale-record.js";
 
 type DynamoDBItem = {
 	tag?: { S: string };
 	path?: { S: string };
 	revalidatedAt?: { N: string };
+	stale?: { N: string };
+	expire?: { N: string };
 };
 
+/**
+ * Determines whether lightweight DynamoDB tag records require blocking regeneration.
+ *
+ * @param items Tag records newer than the cached entry.
+ * @param lastModified Cached entry timestamp.
+ * @param now Current timestamp.
+ * @return Whether a tag was hard-revalidated or its SWR window expired.
+ */
+export function hasHardRevalidation(items: DynamoDBItem[], lastModified: number, now: number): boolean {
+	return items.some((item) => {
+		const revalidatedAt = Number.parseInt(item.revalidatedAt?.N ?? "0");
+		if (revalidatedAt <= lastModified) {
+			return false;
+		}
+		if (item.expire?.N) {
+			const expiry = Number.parseInt(item.expire.N);
+			return expiry <= now && expiry > lastModified;
+		}
+		return !item.stale?.N;
+	});
+}
+
 type DynamoDBResponse = {
 	Items?: DynamoDBItem[];
 };
@@ -56,11 +81,19 @@ function buildDynamoKey(key: string) {
 	return path.posix.join(NEXT_BUILD_ID ?? "", key);
 }
 
-function buildDynamoObject(path: string, tags: string, revalidatedAt?: number) {
+function buildDynamoObject(
+	path: string,
+	tags: string,
+	revalidatedAt?: number,
+	stale?: number,
+	expire?: number
+) {
 	return {
 		path: { S: buildDynamoKey(path) },
 		tag: { S: buildDynamoKey(tags) },
 		revalidatedAt: { N: `${revalidatedAt ?? Date.now()}` },
+		...(stale !== undefined ? { stale: { N: `${stale}` } } : {}),
+		...(expire !== undefined ? { expire: { N: `${expire}` } } : {}),
 	};
 }
 
@@ -72,6 +105,11 @@ const tagCache: OriginalTagCache = {
 				return [];
 			}
 			const { CACHE_DYNAMO_TABLE, NEXT_BUILD_ID } = process.env;
+			const store = globalThis.__openNextAls.getStore();
+			const cache = store?.requestCache.getOrCreate<string, string[]>("dynamoDb:getByPath");
+			if (cache?.has(path)) {
+				return cache.get(path)!;
+			}
 			const result = await awsFetch(
 				JSON.stringify({
 					TableName: CACHE_DYNAMO_TABLE,
@@ -93,7 +131,9 @@ const tagCache: OriginalTagCache = {
 			const tags = Items?.map((item) => item.tag?.S ?? "") ?? [];
 			debug("tags for path", path, tags);
 			// We need to remove the buildId from the path
-			return tags.map((tag: string) => tag.replace(`${NEXT_BUILD_ID}/`, ""));
+			const resultTags = tags.map((tag: string) => tag.replace(`${NEXT_BUILD_ID}/`, ""));
+			cache?.set(path, resultTags);
+			return resultTags;
 		} catch (e) {
 			error("Failed to get tags by path", e);
 			return [];
@@ -105,6 +145,11 @@ const tagCache: OriginalTagCache = {
 				return [];
 			}
 			const { CACHE_DYNAMO_TABLE, NEXT_BUILD_ID } = process.env;
+			const store = globalThis.__openNextAls.getStore();
+			const cache = store?.requestCache.getOrCreate<string, string[]>("dynamoDb:getByTag");
+			if (cache?.has(tag)) {
+				return cache.get(tag)!;
+			}
 			const result = await awsFetch(
 				JSON.stringify({
 					TableName: CACHE_DYNAMO_TABLE,
@@ -121,21 +166,35 @@ const tagCache: OriginalTagCache = {
 				throw new RecoverableError(`Failed to get by tag: ${result.status}`);
 			}
 			const { Items } = (await result.json()) as DynamoDBResponse;
-			return (
-				// We need to remove the buildId from the path
-				Items?.map((item) => item.path?.S?.replace(`${NEXT_BUILD_ID}/`, "") ?? "") ?? []
-			);
+			const paths = Items?.map((item) => item.path?.S?.replace(`${NEXT_BUILD_ID}/`, "") ?? "") ?? [];
+			cache?.set(tag, paths);
+			return paths;
 		} catch (e) {
 			error("Failed to get by tag", e);
 			return [];
 		}
 	},
+	/**
+	 * Checks hard invalidation and shares queried records with subsequent stale checks.
+	 *
+	 * @param key Entry key.
+	 * @param lastModified Entry generation time.
+	 * @return -1 for hard invalidation, otherwise the generation time; preserves it on failure.
+	 */
 	async getLastModified(key: string, lastModified?: number) {
 		try {
 			if (globalThis.openNextConfig.dangerous?.disableTagCache) {
 				return lastModified ?? Date.now();
 			}
 			const { CACHE_DYNAMO_TABLE } = process.env;
+			const store = globalThis.__openNextAls.getStore();
+			const cache = store?.requestCache.getOrCreate<string, DynamoDBItem[]>("dynamoDb:revalidateQueryItems");
+			const cacheKey = `${key}:${lastModified ?? 0}`;
+			if (cache?.has(cacheKey)) {
+				return hasHardRevalidation(cache.get(cacheKey)!, lastModified ?? 0, Date.now())
+					? -1
+					: (lastModified ?? Date.now());
+			}
 			const result = await awsFetch(
 				JSON.stringify({
 					TableName: CACHE_DYNAMO_TABLE,
@@ -156,14 +215,70 @@ const tagCache: OriginalTagCache = {
 			}
 			const revalidatedTags = ((await result.json()) as DynamoDBResponse).Items ?? [];
 			debug("revalidatedTags", revalidatedTags);
-			// If we have revalidated tags we return -1 to force revalidation
-			return revalidatedTags.length > 0 ? -1 : (lastModified ?? Date.now());
+
+			const resultValue = hasHardRevalidation(revalidatedTags, lastModified ?? 0, Date.now())
+				? -1
+				: (lastModified ?? Date.now());
+			cache?.set(cacheKey, revalidatedTags);
+			return resultValue;
 		} catch (e) {
 			error("Failed to get revalidated tags", e);
 			return lastModified ?? Date.now();
 		}
 	},
-	async writeTags(tags: { tag: string; path: string; revalidatedAt?: number }[]) {
+	/**
+	 * Checks applicable stale timestamps rather than treating every queried tag as stale.
+	 *
+	 * @param key Entry key.
+	 * @param lastModified Entry generation time.
+	 * @return Whether an active stale window applies; false on read failure.
+	 */
+	async isStale(key: string, lastModified?: number) {
+		try {
+			if (globalThis.openNextConfig.dangerous?.disableTagCache) {
+				return false;
+			}
+			const { CACHE_DYNAMO_TABLE } = process.env;
+			const store = globalThis.__openNextAls.getStore();
+			const itemsCache = store?.requestCache.getOrCreate<string, DynamoDBItem[]>(
+				"dynamoDb:revalidateQueryItems"
+			);
+			const cacheKey = `${key}:${lastModified ?? 0}`;
+			let items: DynamoDBItem[];
+			if (itemsCache?.has(cacheKey)) {
+				items = itemsCache.get(cacheKey)!;
+			} else {
+				// Share the query with getLastModified, then evaluate stale timestamps separately.
+				const result = await awsFetch(
+					JSON.stringify({
+						TableName: CACHE_DYNAMO_TABLE,
+						IndexName: "revalidate",
+						KeyConditionExpression: "#key = :key AND #revalidatedAt > :lastModified",
+						ExpressionAttributeNames: {
+							"#key": "path",
+							"#revalidatedAt": "revalidatedAt",
+						},
+						ExpressionAttributeValues: {
+							":key": { S: buildDynamoKey(key) },
+							":lastModified": { N: String(lastModified ?? 0) },
+						},
+					})
+				);
+				if (result.status !== 200) {
+					throw new RecoverableError(`Failed to check stale tags: ${result.status}`);
+				}
+				items = ((await result.json()) as DynamoDBResponse).Items ?? [];
+				itemsCache?.set(cacheKey, items);
+			}
+			debug("isStale items", key, items);
+			const now = Date.now();
+			return items.some((item) => isStaleRecord(item, lastModified ?? 0, now));
+		} catch (e) {
+			error("Failed to check stale tags", e);
+			return false;
+		}
+	},
+	async writeTags(tags) {
 		try {
 			const { CACHE_DYNAMO_TABLE } = process.env;
 			if (globalThis.openNextConfig.dangerous?.disableTagCache) {
@@ -174,7 +289,7 @@ const tagCache: OriginalTagCache = {
 					[CACHE_DYNAMO_TABLE ?? ""]: Items.map((Item) => ({
 						PutRequest: {
 							Item: {
-								...buildDynamoObject(Item.path, Item.tag, Item.revalidatedAt),
+								...buildDynamoObject(Item.path, Item.tag, Item.revalidatedAt, Item.stale, Item.expire),
 							},
 						},
 					})),