Merge pull request '[PR]: perf: enforce redis with TTL to cache' (#10397) from nmkj/calckey:replace-cache into develop

Reviewed-on: https://codeberg.org/calckey/calckey/pulls/10397

packages/backend/package.json

@@ -34,6 +34,7 @@
 		"@koa/cors": "3.4.3",
 		"@koa/multer": "3.0.2",
 		"@koa/router": "9.0.1",
+		"@msgpack/msgpack": "3.0.0-beta2",
 		"@peertube/http-signature": "1.7.0",
 		"@redocly/openapi-core": "1.0.0-beta.120",
 		"@sinonjs/fake-timers": "9.1.2",
@@ -43,7 +44,6 @@
 		"ajv": "8.12.0",
 		"archiver": "5.3.1",
 		"argon2": "^0.30.3",
-		"async-mutex": "^0.4.0",
 		"autobind-decorator": "2.4.0",
 		"autolinker": "4.0.0",
 		"autwh": "0.1.0",

packages/backend/src/misc/cache.ts

@@ -1,43 +1,85 @@
+import { redisClient } from "@/db/redis.js";
+import { encode, decode } from "@msgpack/msgpack";
+import { ChainableCommander } from "ioredis";
+
 export class Cache<T> {
-	public cache: Map<string | null, { date: number; value: T }>;
+	private ttl: number;
-	private lifetime: number;
+	private prefix: string;
 
-	constructor(lifetime: Cache<never>["lifetime"]) {
+	constructor(name: string, ttlSeconds: number) {
-		this.cache = new Map();
+		this.ttl = ttlSeconds;
-		this.lifetime = lifetime;
+		this.prefix = `cache:${name}`;
 	}
 
-	public set(key: string | null, value: T): void {
+	private prefixedKey(key: string | null): string {
-		this.cache.set(key, {
+		return key ? `${this.prefix}:${key}` : this.prefix;
-			date: Date.now(),
-			value,
-		});
 	}
 
-	public get(key: string | null): T | undefined {
+	public async set(
-		const cached = this.cache.get(key);
+		key: string | null,
-		if (cached == null) return undefined;
+		value: T,
-		if (Date.now() - cached.date > this.lifetime) {
+		transaction?: ChainableCommander,
-			this.cache.delete(key);
+	): Promise<void> {
-			return undefined;
+		const _key = this.prefixedKey(key);
+		const _value = Buffer.from(encode(value));
+		const commander = transaction ?? redisClient;
+		await commander.set(_key, _value, "EX", this.ttl);
+	}
+
+	public async get(key: string | null, renew = false): Promise<T | undefined> {
+		const _key = this.prefixedKey(key);
+		const cached = await redisClient.getBuffer(_key);
+		if (cached === null) return undefined;
+
+		if (renew) await redisClient.expire(_key, this.ttl);
+
+		return decode(cached) as T;
+	}
+
+	public async getAll(renew = false): Promise<Map<string, T>> {
+		const keys = await redisClient.keys(`${this.prefix}*`);
+		const map = new Map<string, T>();
+		if (keys.length === 0) {
+			return map;
 		}
-		return cached.value;
+		const values = await redisClient.mgetBuffer(keys);
+
+		for (const [i, key] of keys.entries()) {
+			const val = values[i];
+			if (val !== null) {
+				map.set(key, decode(val) as T);
+			}
+		}
+
+		if (renew) {
+			const trans = redisClient.multi();
+			for (const key of map.keys()) {
+				trans.expire(key, this.ttl);
+			}
+			await trans.exec();
+		}
+
+		return map;
 	}
 
-	public delete(key: string | null) {
+	public async delete(...keys: (string | null)[]): Promise<void> {
-		this.cache.delete(key);
+		if (keys.length > 0) {
+			const _keys = keys.map(this.prefixedKey);
+			await redisClient.del(_keys);
+		}
 	}
 
 	/**
-	 * キャッシュがあればそれを返し、無ければfetcherを呼び出して結果をキャッシュ&返します
+	 * Returns if cached value exists. Otherwise, calls fetcher and caches.
-	 * optional: キャッシュが存在してもvalidatorでfalseを返すとキャッシュ無効扱いにします
+	 * Overwrites cached value if invalidated by the optional validator.
 	 */
 	public async fetch(
 		key: string | null,
 		fetcher: () => Promise<T>,
+		renew = false,
 		validator?: (cachedValue: T) => boolean,
 	): Promise<T> {
-		const cachedValue = this.get(key);
+		const cachedValue = await this.get(key, renew);
 		if (cachedValue !== undefined) {
 			if (validator) {
 				if (validator(cachedValue)) {
@@ -52,20 +94,21 @@ export class Cache<T> {
 
 		// Cache MISS
 		const value = await fetcher();
-		this.set(key, value);
+		await this.set(key, value);
 		return value;
 	}
 
 	/**
-	 * キャッシュがあればそれを返し、無ければfetcherを呼び出して結果をキャッシュ&返します
+	 * Returns if cached value exists. Otherwise, calls fetcher and caches if the fetcher returns a value.
-	 * optional: キャッシュが存在してもvalidatorでfalseを返すとキャッシュ無効扱いにします
+	 * Overwrites cached value if invalidated by the optional validator.
 	 */
 	public async fetchMaybe(
 		key: string | null,
 		fetcher: () => Promise<T | undefined>,
+		renew = false,
 		validator?: (cachedValue: T) => boolean,
 	): Promise<T | undefined> {
-		const cachedValue = this.get(key);
+		const cachedValue = await this.get(key, renew);
 		if (cachedValue !== undefined) {
 			if (validator) {
 				if (validator(cachedValue)) {
@@ -81,7 +124,7 @@ export class Cache<T> {
 		// Cache MISS
 		const value = await fetcher();
 		if (value !== undefined) {
-			this.set(key, value);
+			await this.set(key, value);
 		}
 		return value;
 	}

packages/backend/src/misc/check-hit-antenna.ts

@@ -11,7 +11,7 @@ import * as Acct from "@/misc/acct.js";
 import type { Packed } from "./schema.js";
 import { Cache } from "./cache.js";
 
-const blockingCache = new Cache<User["id"][]>(1000 * 60 * 5);
+const blockingCache = new Cache<User["id"][]>("blocking", 60 * 5);
 
 // NOTE: フォローしているユーザーのノート、リストのユーザーのノート、グループのユーザーのノート指定はパフォーマンス上の理由で無効になっている
 

packages/backend/src/misc/emoji-meta.ts

@@ -1,33 +1,41 @@
 import probeImageSize from "probe-image-size";
-import { Mutex, withTimeout } from "async-mutex";
+import { Mutex } from "redis-semaphore";
 
 import { FILE_TYPE_BROWSERSAFE } from "@/const.js";
 import Logger from "@/services/logger.js";
 import { Cache } from "./cache.js";
+import { redisClient } from "@/db/redis.js";
 
 export type Size = {
 	width: number;
 	height: number;
 };
 
-const cache = new Cache<boolean>(1000 * 60 * 10); // once every 10 minutes for the same url
+const cache = new Cache<boolean>("emojiMeta", 60 * 10); // once every 10 minutes for the same url
-const mutex = withTimeout(new Mutex(), 1000);
+const logger = new Logger("emoji");
 
 export async function getEmojiSize(url: string): Promise<Size> {
-	const logger = new Logger("emoji");
+	let attempted = true;
 
-	await mutex.runExclusive(() => {
+	const lock = new Mutex(redisClient, "getEmojiSize");
-		const attempted = cache.get(url);
+	await lock.acquire();
-		if (!attempted) {
-			cache.set(url, true);
-		} else {
-			logger.warn(`Attempt limit exceeded: ${url}`);
-			throw new Error("Too many attempts");
-		}
-	});
 
 	try {
-		logger.info(`Retrieving emoji size from ${url}`);
+		attempted = (await cache.get(url)) === true;
+		if (!attempted) {
+			await cache.set(url, true);
+		}
+	} finally {
+		await lock.release();
+	}
+
+	if (attempted) {
+		logger.warn(`Attempt limit exceeded: ${url}`);
+		throw new Error("Too many attempts");
+	}
+
+	try {
+		logger.debug(`Retrieving emoji size from ${url}`);
 		const { width, height, mime } = await probeImageSize(url, {
 			timeout: 5000,
 		});

packages/backend/src/misc/keypair-store.ts

@@ -3,10 +3,12 @@ import type { User } from "@/models/entities/user.js";
 import type { UserKeypair } from "@/models/entities/user-keypair.js";
 import { Cache } from "./cache.js";
 
-const cache = new Cache<UserKeypair>(Infinity);
+const cache = new Cache<UserKeypair>("keypairStore", 60 * 30);
 
 export async function getUserKeypair(userId: User["id"]): Promise<UserKeypair> {
-	return await cache.fetch(userId, () =>
+	return await cache.fetch(
-		UserKeypairs.findOneByOrFail({ userId: userId }),
+		userId,
+		() => UserKeypairs.findOneByOrFail({ userId: userId }),
+		true,
 	);
 }

packages/backend/src/misc/populate-emojis.ts

@@ -7,8 +7,9 @@ import { isSelfHost, toPunyNullable } from "./convert-host.js";
 import { decodeReaction } from "./reaction-lib.js";
 import config from "@/config/index.js";
 import { query } from "@/prelude/url.js";
+import { redisClient } from "@/db/redis.js";
 
-const cache = new Cache<Emoji | null>(1000 * 60 * 60 * 12);
+const cache = new Cache<Emoji | null>("populateEmojis", 60 * 60 * 12);
 
 /**
  * 添付用絵文字情報
@@ -75,7 +76,7 @@ export async function populateEmoji(
 
 	if (emoji && !(emoji.width && emoji.height)) {
 		emoji = await queryOrNull();
-		cache.set(cacheKey, emoji);
+		await cache.set(cacheKey, emoji);
 	}
 
 	if (emoji == null) return null;
@@ -150,7 +151,7 @@ export async function prefetchEmojis(
 	emojis: { name: string; host: string | null }[],
 ): Promise<void> {
 	const notCachedEmojis = emojis.filter(
-		(emoji) => cache.get(`${emoji.name} ${emoji.host}`) == null,
+		async (emoji) => !(await cache.get(`${emoji.name} ${emoji.host}`)),
 	);
 	const emojisQuery: any[] = [];
 	const hosts = new Set(notCachedEmojis.map((e) => e.host));
@@ -169,7 +170,9 @@ export async function prefetchEmojis(
 					select: ["name", "host", "originalUrl", "publicUrl"],
 			  })
 			: [];
+	const trans = redisClient.multi();
 	for (const emoji of _emojis) {
-		cache.set(`${emoji.name} ${emoji.host}`, emoji);
+		cache.set(`${emoji.name} ${emoji.host}`, emoji, trans);
 	}
+	await trans.exec();
 }

packages/backend/src/models/repositories/user.ts

@@ -1,4 +1,3 @@
-import { URL } from "url";
 import { In, Not } from "typeorm";
 import Ajv from "ajv";
 import type { ILocalUser, IRemoteUser } from "@/models/entities/user.js";
@@ -40,7 +39,10 @@ import {
 } from "../index.js";
 import type { Instance } from "../entities/instance.js";
 
-const userInstanceCache = new Cache<Instance | null>(1000 * 60 * 60 * 3);
+const userInstanceCache = new Cache<Instance | null>(
+	"userInstance",
+	60 * 60 * 3,
+);
 
 type IsUserDetailed<Detailed extends boolean> = Detailed extends true
 	? Packed<"UserDetailed">

packages/backend/src/remote/activitypub/db-resolver.ts

@@ -5,7 +5,6 @@ import type {
 	CacheableRemoteUser,
 	CacheableUser,
 } from "@/models/entities/user.js";
-import { User, IRemoteUser } from "@/models/entities/user.js";
 import type { UserPublickey } from "@/models/entities/user-publickey.js";
 import type { MessagingMessage } from "@/models/entities/messaging-message.js";
 import {
@@ -20,8 +19,11 @@ import type { IObject } from "./type.js";
 import { getApId } from "./type.js";
 import { resolvePerson } from "./models/person.js";
 
-const publicKeyCache = new Cache<UserPublickey | null>(Infinity);
+const publicKeyCache = new Cache<UserPublickey | null>("publicKey", 60 * 30);
-const publicKeyByUserIdCache = new Cache<UserPublickey | null>(Infinity);
+const publicKeyByUserIdCache = new Cache<UserPublickey | null>(
+	"publicKeyByUserId",
+	60 * 30,
+);
 
 export type UriParseResult =
 	| {
@@ -123,17 +125,23 @@ export default class DbResolver {
 			if (parsed.type !== "users") return null;
 
 			return (
-				(await userByIdCache.fetchMaybe(parsed.id, () =>
+				(await userByIdCache.fetchMaybe(
-					Users.findOneBy({
+					parsed.id,
-						id: parsed.id,
+					() =>
-					}).then((x) => x ?? undefined),
+						Users.findOneBy({
+							id: parsed.id,
+						}).then((x) => x ?? undefined),
+					true,
 				)) ?? null
 			);
 		} else {
-			return await uriPersonCache.fetch(parsed.uri, () =>
+			return await uriPersonCache.fetch(
-				Users.findOneBy({
+				parsed.uri,
-					uri: parsed.uri,
+				() =>
-				}),
+					Users.findOneBy({
+						uri: parsed.uri,
+					}),
+				true,
 			);
 		}
 	}
@@ -156,14 +164,17 @@ export default class DbResolver {
 
 				return key;
 			},
+			true,
 			(key) => key != null,
 		);
 
 		if (key == null) return null;
 
 		return {
-			user: (await userByIdCache.fetch(key.userId, () =>
+			user: (await userByIdCache.fetch(
-				Users.findOneByOrFail({ id: key.userId }),
+				key.userId,
+				() => Users.findOneByOrFail({ id: key.userId }),
+				true,
 			)) as CacheableRemoteUser,
 			key,
 		};
@@ -183,6 +194,7 @@ export default class DbResolver {
 		const key = await publicKeyByUserIdCache.fetch(
 			user.id,
 			() => UserPublickeys.findOneBy({ userId: user.id }),
+			true,
 			(v) => v != null,
 		);
 

packages/backend/src/remote/activitypub/models/person.ts

@@ -135,14 +135,14 @@ export async function fetchPerson(
 ): Promise<CacheableUser | null> {
 	if (typeof uri !== "string") throw new Error("uri is not string");
 
-	const cached = uriPersonCache.get(uri);
+	const cached = await uriPersonCache.get(uri, true);
 	if (cached) return cached;
 
 	// Fetch from the database if the URI points to this server
 	if (uri.startsWith(`${config.url}/`)) {
 		const id = uri.split("/").pop();
 		const u = await Users.findOneBy({ id });
-		if (u) uriPersonCache.set(uri, u);
+		if (u) await uriPersonCache.set(uri, u);
 		return u;
 	}
 
@@ -150,7 +150,7 @@ export async function fetchPerson(
 	const exist = await Users.findOneBy({ uri });
 
 	if (exist) {
-		uriPersonCache.set(uri, exist);
+		await uriPersonCache.set(uri, exist);
 		return exist;
 	}
 	//#endregion

packages/backend/src/server/api/authenticate.ts

@@ -9,7 +9,7 @@ import {
 	localUserByNativeTokenCache,
 } from "@/services/user-cache.js";
 
-const appCache = new Cache<App>(Infinity);
+const appCache = new Cache<App>("app", 60 * 30);
 
 export class AuthenticationError extends Error {
 	constructor(message: string) {
@@ -49,6 +49,7 @@ export default async (
 		const user = await localUserByNativeTokenCache.fetch(
 			token,
 			() => Users.findOneBy({ token }) as Promise<ILocalUser | null>,
+			true,
 		);
 
 		if (user == null) {
@@ -82,11 +83,14 @@ export default async (
 				Users.findOneBy({
 					id: accessToken.userId,
 				}) as Promise<ILocalUser>,
+			true,
 		);
 
 		if (accessToken.appId) {
-			const app = await appCache.fetch(accessToken.appId, () =>
+			const app = await appCache.fetch(
-				Apps.findOneByOrFail({ id: accessToken.appId! }),
+				accessToken.appId,
+				() => Apps.findOneByOrFail({ id: accessToken.appId! }),
+				true,
 			);
 
 			return [

packages/backend/src/server/api/endpoints/admin/emoji/add.ts

@@ -6,7 +6,7 @@ import { ApiError } from "../../../error.js";
 import rndstr from "rndstr";
 import { publishBroadcastStream } from "@/services/stream.js";
 import { db } from "@/db/postgre.js";
-import { type Size, getEmojiSize } from "@/misc/emoji-meta.js";
+import { getEmojiSize } from "@/misc/emoji-meta.js";
 
 export const meta = {
 	tags: ["admin"],
@@ -40,12 +40,7 @@ export default define(meta, paramDef, async (ps, me) => {
 		? file.name.split(".")[0]
 		: `_${rndstr("a-z0-9", 8)}_`;
 
-	let size: Size = { width: 0, height: 0 };
+	const size = await getEmojiSize(file.url);
-	try {
-		size = await getEmojiSize(file.url);
-	} catch {
-		/* skip if any error happens */
-	}
 
 	const emoji = await Emojis.insert({
 		id: genId(),

packages/backend/src/server/api/endpoints/admin/emoji/copy.ts

@@ -6,7 +6,7 @@ import type { DriveFile } from "@/models/entities/drive-file.js";
 import { uploadFromUrl } from "@/services/drive/upload-from-url.js";
 import { publishBroadcastStream } from "@/services/stream.js";
 import { db } from "@/db/postgre.js";
-import { type Size, getEmojiSize } from "@/misc/emoji-meta.js";
+import { getEmojiSize } from "@/misc/emoji-meta.js";
 
 export const meta = {
 	tags: ["admin"],
@@ -65,12 +65,7 @@ export default define(meta, paramDef, async (ps, me) => {
 		throw new ApiError();
 	}
 
-	let size: Size = { width: 0, height: 0 };
+	const size = await getEmojiSize(driveFile.url);
-	try {
-		size = await getEmojiSize(driveFile.url);
-	} catch {
-		/* skip if any error happens */
-	}
 
 	const copied = await Emojis.insert({
 		id: genId(),

packages/backend/src/server/nodeinfo.ts

@@ -100,7 +100,10 @@ const nodeinfo2 = async () => {
 	};
 };
 
-const cache = new Cache<Awaited<ReturnType<typeof nodeinfo2>>>(1000 * 60 * 10);
+const cache = new Cache<Awaited<ReturnType<typeof nodeinfo2>>>(
+	"nodeinfo",
+	60 * 10,
+);
 
 router.get(nodeinfo2_1path, async (ctx) => {
 	const base = await cache.fetch(null, () => nodeinfo2());

packages/backend/src/services/chart/charts/active-users.ts

@@ -25,12 +25,12 @@ export default class ActiveUsersChart extends Chart<typeof schema> {
 		return {};
 	}
 
-	public async read(user: {
+	public read(user: {
 		id: User["id"];
 		host: null;
 		createdAt: User["createdAt"];
-	}): Promise<void> {
+	})  {
-		await this.commit({
+		this.commit({
 			read: [user.id],
 			registeredWithinWeek:
 				Date.now() - user.createdAt.getTime() < week ? [user.id] : [],

packages/backend/src/services/instance-actor.ts

@@ -6,10 +6,10 @@ import { IsNull } from "typeorm";
 
 const ACTOR_USERNAME = "instance.actor" as const;
 
-const cache = new Cache<ILocalUser>(Infinity);
+const cache = new Cache<ILocalUser>("instanceActor", 60 * 30);
 
 export async function getInstanceActor(): Promise<ILocalUser> {
-	const cached = cache.get(null);
+	const cached = await cache.get(null, true);
 	if (cached) return cached;
 
 	const user = (await Users.findOneBy({
@@ -18,11 +18,11 @@ export async function getInstanceActor(): Promise<ILocalUser> {
 	})) as ILocalUser | undefined;
 
 	if (user) {
-		cache.set(null, user);
+		await cache.set(null, user);
 		return user;
 	} else {
 		const created = (await createSystemUser(ACTOR_USERNAME)) as ILocalUser;
-		cache.set(null, created);
+		await cache.set(null, created);
 		return created;
 	}
 }

packages/backend/src/services/note/create.ts

@@ -29,17 +29,14 @@ import {
 	Notes,
 	Instances,
 	UserProfiles,
-	Antennas,
-	Followings,
 	MutedNotes,
 	Channels,
 	ChannelFollowings,
-	Blockings,
 	NoteThreadMutings,
 } from "@/models/index.js";
 import type { DriveFile } from "@/models/entities/drive-file.js";
 import type { App } from "@/models/entities/app.js";
-import { Not, In, IsNull } from "typeorm";
+import { Not, In } from "typeorm";
 import type { User, ILocalUser, IRemoteUser } from "@/models/entities/user.js";
 import { genId } from "@/misc/gen-id.js";
 import {
@@ -73,7 +70,7 @@ import { Mutex } from "redis-semaphore";
 
 const mutedWordsCache = new Cache<
 	{ userId: UserProfile["userId"]; mutedWords: UserProfile["mutedWords"] }[]
->(1000 * 60 * 5);
+>("mutedWords", 60 * 5);
 
 type NotificationType = "reply" | "renote" | "quote" | "mention";
 

packages/backend/src/services/register-or-fetch-instance-doc.ts

@@ -4,30 +4,30 @@ import { genId } from "@/misc/gen-id.js";
 import { toPuny } from "@/misc/convert-host.js";
 import { Cache } from "@/misc/cache.js";
 
-const cache = new Cache<Instance>(1000 * 60 * 60);
+const cache = new Cache<Instance>("registerOrFetchInstanceDoc", 60 * 60);
 
 export async function registerOrFetchInstanceDoc(
 	host: string,
 ): Promise<Instance> {
-	host = toPuny(host);
+	const _host = toPuny(host);
 
-	const cached = cache.get(host);
+	const cached = await cache.get(_host);
 	if (cached) return cached;
 
-	const index = await Instances.findOneBy({ host });
+	const index = await Instances.findOneBy({ host: _host });
 
 	if (index == null) {
 		const i = await Instances.insert({
 			id: genId(),
-			host,
+			host: _host,
 			caughtAt: new Date(),
 			lastCommunicatedAt: new Date(),
 		}).then((x) => Instances.findOneByOrFail(x.identifiers[0]));
 
-		cache.set(host, i);
+		await cache.set(_host, i);
 		return i;
 	} else {
-		cache.set(host, index);
+		await cache.set(_host, index);
 		return index;
 	}
 }

packages/backend/src/services/relay.ts

@@ -15,7 +15,7 @@ import { createSystemUser } from "./create-system-user.js";
 
 const ACTOR_USERNAME = "relay.actor" as const;
 
-const relaysCache = new Cache<Relay[]>(1000 * 60 * 10);
+const relaysCache = new Cache<Relay[]>("relay", 60 * 10);
 
 export async function getRelayActor(): Promise<ILocalUser> {
 	const user = await Users.findOneBy({
@@ -90,7 +90,7 @@ async function updateRelaysCache() {
 	const relays = await Relays.findBy({
 		status: "accepted",
 	});
-	relaysCache.set(null, relays);
+	await relaysCache.set(null, relays);
 }
 
 export async function relayRejected(id: string) {

packages/backend/src/services/user-cache.ts

@@ -3,17 +3,23 @@ import type {
 	CacheableUser,
 	ILocalUser,
 } from "@/models/entities/user.js";
-import { User } from "@/models/entities/user.js";
 import { Users } from "@/models/index.js";
 import { Cache } from "@/misc/cache.js";
-import { subscriber } from "@/db/redis.js";
+import { redisClient, subscriber } from "@/db/redis.js";
 
-export const userByIdCache = new Cache<CacheableUser>(Infinity);
+export const userByIdCache = new Cache<CacheableUser>("userById", 60 * 30);
 export const localUserByNativeTokenCache = new Cache<CacheableLocalUser | null>(
-	Infinity,
+	"localUserByNativeToken",
+	60 * 30,
+);
+export const localUserByIdCache = new Cache<CacheableLocalUser>(
+	"localUserByIdCache",
+	60 * 30,
+);
+export const uriPersonCache = new Cache<CacheableUser | null>(
+	"uriPerson",
+	60 * 30,
 );
-export const localUserByIdCache = new Cache<CacheableLocalUser>(Infinity);
-export const uriPersonCache = new Cache<CacheableUser | null>(Infinity);
 
 subscriber.on("message", async (_, data) => {
 	const obj = JSON.parse(data);
@@ -22,13 +28,12 @@ subscriber.on("message", async (_, data) => {
 		const { type, body } = obj.message;
 		switch (type) {
 			case "localUserUpdated": {
-				userByIdCache.delete(body.id);
+				await userByIdCache.delete(body.id);
-				localUserByIdCache.delete(body.id);
+				await localUserByIdCache.delete(body.id);
-				localUserByNativeTokenCache.cache.forEach((v, k) => {
+				const toDelete = Array.from(await localUserByNativeTokenCache.getAll())
-					if (v.value?.id === body.id) {
+					.filter((v) => v[1]?.id === body.id)
-						localUserByNativeTokenCache.delete(k);
+					.map((v) => v[0]);
-					}
+				await localUserByNativeTokenCache.delete(...toDelete);
-				});
 				break;
 			}
 			case "userChangeSuspendedState":
@@ -36,15 +41,17 @@ subscriber.on("message", async (_, data) => {
 			case "userChangeModeratorState":
 			case "remoteUserUpdated": {
 				const user = await Users.findOneByOrFail({ id: body.id });
-				userByIdCache.set(user.id, user);
+				await userByIdCache.set(user.id, user);
-				for (const [k, v] of uriPersonCache.cache.entries()) {
+				const trans = redisClient.multi();
-					if (v.value?.id === user.id) {
+				for (const [k, v] of (await uriPersonCache.getAll()).entries()) {
-						uriPersonCache.set(k, user);
+					if (v?.id === user.id) {
+						await uriPersonCache.set(k, user, trans);
 					}
 				}
+				await trans.exec();
 				if (Users.isLocalUser(user)) {
-					localUserByNativeTokenCache.set(user.token, user);
+					await localUserByNativeTokenCache.set(user.token, user);
-					localUserByIdCache.set(user.id, user);
+					await localUserByIdCache.set(user.id, user);
 				}
 				break;
 			}
@@ -52,8 +59,8 @@ subscriber.on("message", async (_, data) => {
 				const user = (await Users.findOneByOrFail({
 					id: body.id,
 				})) as ILocalUser;
-				localUserByNativeTokenCache.delete(body.oldToken);
+				await localUserByNativeTokenCache.delete(body.oldToken);
-				localUserByNativeTokenCache.set(body.newToken, user);
+				await localUserByNativeTokenCache.set(body.newToken, user);
 				break;
 			}
 			default:

pnpm-lock.yaml

@@ -105,6 +105,9 @@ importers:
       '@koa/router':
         specifier: 9.0.1
         version: 9.0.1
+      '@msgpack/msgpack':
+        specifier: 3.0.0-beta2
+        version: 3.0.0-beta2
       '@peertube/http-signature':
         specifier: 1.7.0
         version: 1.7.0
@@ -132,9 +135,6 @@ importers:
       argon2:
         specifier: ^0.30.3
         version: 0.30.3
-      async-mutex:
-        specifier: ^0.4.0
-        version: 0.4.0
       autobind-decorator:
         specifier: 2.4.0
         version: 2.4.0
@@ -786,7 +786,7 @@ importers:
         version: 2.30.0
       emojilib:
         specifier: github:thatonecalculator/emojilib
-        version: github.com/thatonecalculator/emojilib/542fcc1a25003afad78f3248ceee8ac6980ddeb8
+        version: github.com/thatonecalculator/emojilib/06944984a61ee799b7083894258f5fa318d932d1
       escape-regexp:
         specifier: 0.0.1
         version: 0.0.1
@@ -2277,6 +2277,11 @@ packages:
       os-filter-obj: 2.0.0
     dev: true
 
+  /@msgpack/msgpack@3.0.0-beta2:
+    resolution: {integrity: sha512-y+l1PNV0XDyY8sM3YtuMLK5vE3/hkfId+Do8pLo/OPxfxuFAUwcGz3oiiUuV46/aBpwTzZ+mRWVMtlSKbradhw==}
+    engines: {node: '>= 14'}
+    dev: false
+
   /@msgpackr-extract/msgpackr-extract-darwin-arm64@3.0.2:
     resolution: {integrity: sha512-9bfjwDxIDWmmOKusUcqdS4Rw+SETlp9Dy39Xui9BEGEk19dDwH0jhipwFzEff/pFg95NKymc6TOTbRKcWeRqyQ==}
     cpu: [arm64]
@@ -4496,12 +4501,6 @@ packages:
       stream-exhaust: 1.0.2
     dev: true
 
-  /async-mutex@0.4.0:
-    resolution: {integrity: sha512-eJFZ1YhRR8UN8eBLoNzcDPcy/jqjsg6I1AP+KvWQX80BqOSW1oJPJXDylPUEeMr2ZQvHgnQ//Lp6f3RQ1zI7HA==}
-    dependencies:
-      tslib: 2.6.0
-    dev: false
-
   /async-settle@1.0.0:
     resolution: {integrity: sha512-VPXfB4Vk49z1LHHodrEQ6Xf7W4gg1w0dAPROHngx7qgDjqmIQ+fXmwgGXTW/ITLai0YLSvWepJOP9EVpMnEAcw==}
     engines: {node: '>= 0.10'}
@@ -15772,8 +15771,8 @@ packages:
       url-polyfill: 1.1.12
     dev: true
 
-  github.com/thatonecalculator/emojilib/542fcc1a25003afad78f3248ceee8ac6980ddeb8:
+  github.com/thatonecalculator/emojilib/06944984a61ee799b7083894258f5fa318d932d1:
-    resolution: {tarball: https://codeload.github.com/thatonecalculator/emojilib/tar.gz/542fcc1a25003afad78f3248ceee8ac6980ddeb8}
+    resolution: {tarball: https://codeload.github.com/thatonecalculator/emojilib/tar.gz/06944984a61ee799b7083894258f5fa318d932d1}
     name: emojilib
     version: 3.0.10
     dev: true