From bd5aa901aba9a420c7119f5cfcfe9a70873feed8 Mon Sep 17 00:00:00 2001 From: Juan Castro Date: Tue, 6 Oct 2026 18:55:53 -0400 Subject: [PATCH] perf: batch the per-page reads in a team delete Measured first, against a real server, counting DB round trips: a 40-seat delete was 284 queries, 7.1 per seat, scaling linearly. Four of those seven answered for the whole page and had no reason to be asked per account. 120x SELECT * FROM `user` WHERE `id` = ? 41x INSERT INTO `audit_team_membership` 40x SELECT COALESCE(SUM(size), 0) FROM fsentries 40x UPDATE `user` SET `suspended` = ? 40x SELECT `uuid` FROM `sessions` WHERE `user_id` = ? The page now reads who is already suspended, what each holds, and the rows the notices need in one query each, and writes the audit as one statement. What is genuinely per account -- the suspension, its billing event, the mail -- stays per account under the same concurrency bound. Same 40-seat delete is now 170 queries, 4.3 per seat. Three new store methods rather than raw SQL at the call site: `getSuspendedByIds` reads the primary, deliberately not `getByIds`, which is cache-first. A stale `suspended` is exactly what re-charges a seat on a retry, which is the thing this path was just fixed to stop doing. `getHeldBytesForUsers` is `getHeldBytes` with a GROUP BY. `appendAuditMany` is `appendAudit` as one multi-row insert, still written before any suspension, so a failed append cannot leave an unlogged one. What is left is concentrated in `#suspend`: the UPDATE, the forced re-read the user store does afterwards to refresh its cache, and the session drop. Those are shared store paths and batching them means owning their cache invalidation, so they are not in scope here. --- src/backend/services/team/TeamService.ts | 105 ++++++++++++++--------- src/backend/stores/fs/FSEntryStore.ts | 32 +++++++ src/backend/stores/team/TeamStore.ts | 32 +++++++ src/backend/stores/user/UserStore.ts | 34 ++++++++ 4 files changed, 161 insertions(+), 42 deletions(-) diff --git a/src/backend/services/team/TeamService.ts b/src/backend/services/team/TeamService.ts index d32e8e105..818fc42f9 100644 --- a/src/backend/services/team/TeamService.ts +++ b/src/backend/services/team/TeamService.ts @@ -638,22 +638,12 @@ export class TeamService extends PuterService { // Otherwise they keep working, unreachable through a deleted team. let page = await this.stores.team.listMembers(teamUid, { limit: 200 }); for (;;) { - // Bounded: a seat costs a mail send, and a full team is 40 of them. - const outcomes = await runWithConcurrencyLimitSettled( + disabled += await this.#disableSeatsForDelete( page.items.filter((m) => Number(m.org_owned) === 1), - TEARDOWN_CONCURRENCY, - (member) => - this.#disableSeatForDelete( - member, - team, - actorUserId, - billing, - ), + team, + actorUserId, + billing, ); - disabled += outcomes.filter((o) => o.status === 'fulfilled').length; - // Settled: a rejection cannot cancel its peers, so finish the page. - const failed = outcomes.find((o) => o.status === 'rejected'); - if (failed) throw failed.reason; if (!page.cursor) break; page = await this.stores.team.listMembers(teamUid, { limit: 200, @@ -675,41 +665,72 @@ export class TeamService extends PuterService { }); } - /** As `disableMember`; a suspended seat is skipped so a retry resumes. */ - async #disableSeatForDelete( - member: TeamMemberRow, + /** + * One page of seats, as `disableMember` does one member. Everything that + * answers for the whole page -- who is already off, what they hold, the + * audit -- goes out once; what is genuinely per account stays per account. + * Returns how many of them end up disabled. + */ + async #disableSeatsForDelete( + seats: TeamMemberRow[], team: TeamRow, actorUserId: number, billing: TeamBillingContext, - ): Promise { - const current = await this.stores.user.getByProperty( - 'id', - member.user_id, - { force: true }, + ): Promise { + if (seats.length === 0) return 0; + + // From the primary: a cached `suspended` would re-charge on a retry. + const suspended = await this.stores.user.getSuspendedByIds( + seats.map((m) => m.user_id), ); - if (current?.suspended) return; + const todo = seats.filter((m) => !suspended.get(m.user_id)); + // Already off, and still accounts the team is losing. + if (todo.length === 0) return seats.length; - await this.stores.team.appendAudit({ - teamId: team.id, - userId: member.user_id, - actorUserId, - action: 'disable', - reason: 'team_deleted', - }); - const held = await this.#heldBytes(member.user_id); - await this.#suspend(member.user_id); + const ids = todo.map((m) => m.user_id); + const held = await this.stores.fsEntry.getHeldBytesForUsers(ids); + const rows = await this.stores.user.getByIds(ids); + // Recorded first, so a failed append cannot leave an unlogged suspension. + await this.stores.team.appendAuditMany( + todo.map((m) => ({ + teamId: team.id, + userId: m.user_id, + actorUserId, + action: 'disable', + reason: 'team_deleted', + })), + ); - // Per seat, not one bulk event: the byte charge is per account. - this.#emitBilling('team.account.disabled', { - ...billing, - user_id: member.user_id, - user_uuid: member.uuid, - username: member.username, - held_bytes: held, - }); + // Bounded: what is left per seat is a write and a mail send. + const outcomes = await runWithConcurrencyLimitSettled( + todo, + TEARDOWN_CONCURRENCY, + async (member) => { + await this.#suspend(member.user_id); - // The team notice covers the disabling, so a member is told once. - await this.#notifyMember(member.user_id, 'team_closed', team); + // Per seat, not one bulk event: the byte charge is per account. + this.#emitBilling('team.account.disabled', { + ...billing, + user_id: member.user_id, + user_uuid: member.uuid, + username: member.username, + held_bytes: held.get(member.user_id) ?? 0, + }); + + // The team notice covers the disabling, so a member is told once. + await this.#notifyUser( + rows.get(member.user_id), + 'team_closed', + team, + ); + }, + ); + + const done = outcomes.filter((o) => o.status === 'fulfilled').length; + // Settled: a rejection cannot cancel its peers, so finish the page. + const failed = outcomes.find((o) => o.status === 'rejected'); + if (failed) throw failed.reason; + return seats.length - todo.length + done; } /** Team owner only. Readable after deletion -- that is the point of it. */ diff --git a/src/backend/stores/fs/FSEntryStore.ts b/src/backend/stores/fs/FSEntryStore.ts index 84a81a783..d33985a1c 100644 --- a/src/backend/stores/fs/FSEntryStore.ts +++ b/src/backend/stores/fs/FSEntryStore.ts @@ -3344,6 +3344,38 @@ export class FSEntryStore extends PuterStore { return Number(rows[0]?.totalUsage ?? 0); } + /** + * Held bytes for several accounts in one grouped read, as `getHeldBytes` + * does for one. + */ + async getHeldBytesForUsers( + userIds: number[], + ): Promise> { + const out = new Map(); + const unique = [ + ...new Set( + (Array.isArray(userIds) ? userIds : []).filter( + (id): id is number => typeof id === 'number', + ), + ), + ]; + for (const id of unique) out.set(id, 0); + + for (let offset = 0; offset < unique.length; offset += 200) { + const chunk = unique.slice(offset, offset + 200); + const placeholders = chunk.map(() => '?').join(', '); + const rows = (await this.clients.db.read( + `SELECT user_id, COALESCE(SUM(size), 0) AS ${this.clients.db.quoteIdentifier('totalUsage')} ` + + `FROM fsentries WHERE user_id IN (${placeholders}) GROUP BY user_id`, + chunk, + )) as { user_id: number; totalUsage: number }[]; + for (const row of rows) { + out.set(Number(row.user_id), Number(row.totalUsage ?? 0)); + } + } + return out; + } + /** * `consistentRead` reads both queries from the primary instead of a replica * — used for the exact re-check when a replica-backed read would reject a diff --git a/src/backend/stores/team/TeamStore.ts b/src/backend/stores/team/TeamStore.ts index 4e08f4c67..1cbd46cab 100644 --- a/src/backend/stores/team/TeamStore.ts +++ b/src/backend/stores/team/TeamStore.ts @@ -820,6 +820,38 @@ export class TeamStore extends PuterStore { ); } + /** The same rows as `appendAudit`, in one statement per chunk. */ + async appendAuditMany( + entries: Array<{ + teamId: number; + userId: number; + actorUserId: number; + action: string; + reason?: string | null; + }>, + ): Promise { + if (entries.length === 0) return; + for (let offset = 0; offset < entries.length; offset += 200) { + const chunk = entries.slice(offset, offset + 200); + const values = chunk.map(() => '(?, ?, ?, ?, ?, ?, ?)').join(', '); + const params = chunk.flatMap((e) => [ + e.teamId, + e.teamId, + e.userId, + e.userId, + e.actorUserId, + e.action, + e.reason ?? null, + ]); + await this.clients.db.write( + 'INSERT INTO `audit_team_membership` ' + + '(`group_id`, `group_id_keep`, `user_id`, `user_id_keep`, ' + + `\`actor_user_id\`, \`action\`, \`reason\`) VALUES ${values}`, + params, + ); + } + } + /** The whole team's audit, newest first, keyset-paginated on `id`. */ async listAudit( teamId: number, diff --git a/src/backend/stores/user/UserStore.ts b/src/backend/stores/user/UserStore.ts index 176ff4c9a..c077b25c5 100644 --- a/src/backend/stores/user/UserStore.ts +++ b/src/backend/stores/user/UserStore.ts @@ -248,6 +248,40 @@ export class UserStore extends PuterStore { return this.getByProperty('referral_code', normalized, opts); } + /** + * Which of these accounts are suspended, read from the primary. + * + * Deliberately not `getByIds`, which answers from cache: a caller deciding + * whether to suspend an account needs the row as it is now, not as it was + * when something last cached it. + */ + async getSuspendedByIds(ids: number[]): Promise> { + const out = new Map(); + const unique = [ + ...new Set( + (Array.isArray(ids) ? ids : []).filter( + (id): id is number => typeof id === 'number', + ), + ), + ]; + for ( + let offset = 0; + offset < unique.length; + offset += BULK_QUERY_CHUNK_SIZE + ) { + const chunk = unique.slice(offset, offset + BULK_QUERY_CHUNK_SIZE); + const placeholders = chunk.map(() => '?').join(', '); + const rows = (await this.clients.db.pread( + `SELECT \`id\`, \`suspended\` FROM \`user\` WHERE \`id\` IN (${placeholders})`, + chunk, + )) as Array<{ id: number; suspended: unknown }>; + for (const row of rows) { + out.set(Number(row.id), Boolean(row.suspended)); + } + } + return out; + } + /** * Batched lookup by id. Dedupes input ids, reads cache via a pipelined * MGET, and resolves remaining misses with a single `SELECT … WHERE id IN