diff --git a/src/backend/clients/database/DatabaseClient.test.ts b/src/backend/clients/database/DatabaseClient.test.ts index a90e8f8e4..04ca504a9 100644 --- a/src/backend/clients/database/DatabaseClient.test.ts +++ b/src/backend/clients/database/DatabaseClient.test.ts @@ -275,6 +275,16 @@ describe('AbstractDatabaseClient — SQL generation', () => { expect(postgres().timestampFromUnixSql()).toBe('to_timestamp(?)'); }); + it('reads a timestamp column back as unix seconds in each engine', () => { + expect(sqlite().unixSecondsSql('`t`')).toBe( + "CAST(strftime('%s', `t`) AS INTEGER)", + ); + expect(mysql().unixSecondsSql('`t`')).toBe('UNIX_TIMESTAMP(`t`)'); + expect(postgres().unixSecondsSql('`t`')).toBe( + 'EXTRACT(EPOCH FROM `t`)::bigint', + ); + }); + it('describes a stored timestamp the way each driver reads it back', () => { const seconds = Date.UTC(2026, 9, 11, 8, 5, 9) / 1000; expect(sqlite().timestampValue(seconds)).toBe('2026-10-11 08:05:09'); diff --git a/src/backend/clients/database/DatabaseClient.ts b/src/backend/clients/database/DatabaseClient.ts index 2c93ce5ef..b4ba0b82b 100644 --- a/src/backend/clients/database/DatabaseClient.ts +++ b/src/backend/clients/database/DatabaseClient.ts @@ -388,6 +388,19 @@ export class AbstractDatabaseClient extends PuterClient { }); } + /** + * A timestamp column as unix seconds, computed by the database: the mysql + * driver reads a stored UTC timestamp as local time, so parsing it in JS + * can be off by the server's UTC offset. + */ + unixSecondsSql(column: string): string { + return this.case({ + postgres: `EXTRACT(EPOCH FROM ${column})::bigint`, + mysql: `UNIX_TIMESTAMP(${column})`, + otherwise: `CAST(strftime('%s', ${column}) AS INTEGER)`, + }); + } + /** * What a read of a timestamp column holding `seconds` returns: sqlite gives * back the UTC text it stored, mysql2 and pg a `Date` read in node's time diff --git a/src/backend/controllers/auth/AuthController.ts b/src/backend/controllers/auth/AuthController.ts index 805b32df0..b7208af2b 100644 --- a/src/backend/controllers/auth/AuthController.ts +++ b/src/backend/controllers/auth/AuthController.ts @@ -3836,9 +3836,10 @@ export class AuthController extends PuterController { // An origin's uid is a deterministic uuidv5, so a deleted app // reappears here under the identical uid. Withdraw any cross-app // data grants left pointing at it before this new row can inherit - // consent the user gave its predecessor. Only *this* path can reuse - // a uid: `AppStore.create` mints a random uuid4, which no deleted - // app can ever hold again. + // consent the user gave its predecessor. Only `createFromOrigin` + // reuses a uid, here or wherever else an origin app is bootstrapped: + // `AppStore.create` mints a random uuid4, which no deleted app can + // ever hold again. // // Called directly rather than through `app.changed`: the token is // issued below, so this has to be able to stop that, and @@ -3849,6 +3850,9 @@ export class AuthController extends PuterController { app_uid, 'uid reused by a new app', ); + // Key-value shares the predecessor left read as gone already, + // being older than this row; this only cleans them up. + this.services.kvShare.retireSharesOlderThanApp(app); } if (!app) { throw new HttpError(404, `App ${app_uid} does not exist`, { diff --git a/src/backend/drivers/kv/KVStoreDriver.ts b/src/backend/drivers/kv/KVStoreDriver.ts index 76de6f9a3..f363e0c7c 100644 --- a/src/backend/drivers/kv/KVStoreDriver.ts +++ b/src/backend/drivers/kv/KVStoreDriver.ts @@ -57,6 +57,7 @@ import { sealCursor, unsealCursor, } from '../../util/pagination.js'; +import { predatesApp } from '../../util/appGeneration.js'; import { KV_COSTS } from './costs.js'; import { KV_HANDLE_NAMESPACE_BUDGET, @@ -507,11 +508,15 @@ export class KVStoreDriver extends PuterDriver { ) throw handleBudgetSpent(); - // Side by side, so the app lookup costs no round trip of its own. + // Side by side, so the app lookup costs no round trip of its own. A + // handle older than its app was made for an earlier app under the + // same uid, and is gone with it. const [appGone, allowed] = await Promise.all([ row.appUid === KV_GLOBAL_APP_KEY ? false - : this.stores.app.getByUid(row.appUid).then((app) => !app), + : this.stores.app + .getByUid(row.appUid) + .then((app) => !app || predatesApp(row.createdAt, app)), this.services.permission.check( actor, kvSharePermission(ownerUuid, row.appUid, row.keyPrefix, level), diff --git a/src/backend/services/auth/AuthService.ts b/src/backend/services/auth/AuthService.ts index 106009c22..a9d75ee68 100644 --- a/src/backend/services/auth/AuthService.ts +++ b/src/backend/services/auth/AuthService.ts @@ -28,6 +28,7 @@ import { import { HttpError } from '../../core/http/HttpError.js'; import { WEB_AND_EXTENSION_PROTOCOLS } from '../../util/validation.js'; import { awaitLater } from '../../util/concurrency.js'; +import { appGenerationStart } from '../../util/appGeneration.js'; import { ASSET_WINDOW_SECONDS, GODMODE_TOKEN_MAX_AGE_SECONDS, @@ -57,9 +58,6 @@ import type { const APP_ORIGIN_UUID_NAMESPACE = '33de3768-8ee0-43e9-9e73-db192b97a5d8'; // Successor uids an origin can derive after its earlier apps were repointed. const MAX_ORIGIN_UID_GENERATIONS = 8; -// Allowed gap between a session's creation time (server clock) and its app's -// (database clock) before the session counts as older than the app. -const APP_SESSION_CLOCK_SKEW_SECONDS = 60; // Revoked and expired sessions are kept this long, so a stale token still gets // `session_revoked`/`session_expired` rather than `invalid`, then deleted. Web @@ -562,7 +560,7 @@ export class AuthService extends PuterService { appUid, workerName, auth_id, - notBefore: this.#appSessionsNotBefore(app), + notBefore: appGenerationStart(app), }, ); if (!session) { @@ -1808,7 +1806,7 @@ export class AuthService extends PuterService { appUid, { auth_id: this.#authIdFor(actor.user as UserRow), - notBefore: this.#appSessionsNotBefore(app), + notBefore: appGenerationStart(app), }, ); @@ -2685,7 +2683,7 @@ export class AuthService extends PuterService { // uid returns with the next app on that origin. A session older than // the app was made for the earlier one. Checked before `touch()` so a // rejected token doesn't slide the old session's expiry. - const notBefore = this.#appSessionsNotBefore(app); + const notBefore = appGenerationStart(app); const createdAt = Number(session.created_at); if (notBefore !== null && createdAt > 0 && createdAt < notBefore) { return { reauth: { reason: 'session_revoked' } }; @@ -2714,19 +2712,6 @@ export class AuthService extends PuterService { return block.blocked ? { reason: block.reason } : null; } - /** - * Unix seconds before which an app or worker session for `app` belongs to - * an earlier app with the same uid; null when the creation time is - * unknown. - */ - #appSessionsNotBefore( - app: { created_epoch?: unknown } | null | undefined, - ): number | null { - const created = Number(app?.created_epoch); - if (!Number.isFinite(created) || created <= 0) return null; - return created - APP_SESSION_CLOCK_SKEW_SECONDS; - } - #createdBefore(row: SessionRow | null, notBefore: number | null): boolean { if (notBefore === null || !row) return false; const createdAt = Number(row.created_at); @@ -2802,7 +2787,7 @@ export class AuthService extends PuterService { if (blocked) return { blocked }; // As with app sessions: a token minted under an earlier app that // held this uid doesn't authenticate as the current one. - const notBefore = this.#appSessionsNotBefore(app); + const notBefore = appGenerationStart(app); if ( this.#createdBefore(session, notBefore) || this.#createdBefore(parentSession, notBefore) diff --git a/src/backend/services/events/EventsService.test.ts b/src/backend/services/events/EventsService.test.ts index b08b79950..09b7e2443 100644 --- a/src/backend/services/events/EventsService.test.ts +++ b/src/backend/services/events/EventsService.test.ts @@ -205,6 +205,7 @@ let handles: Map; const kvShareHandleStore = { getByHandle: async (handle: string) => handles.get(handle) ?? null, + predatesItsApp: async () => false, getLive: async (handle: string) => { const row = handles.get(handle); if (!row || row.revokedAt !== null) return null; diff --git a/src/backend/services/events/EventsService.ts b/src/backend/services/events/EventsService.ts index 6b9d72507..8fb8dfec4 100644 --- a/src/backend/services/events/EventsService.ts +++ b/src/backend/services/events/EventsService.ts @@ -3173,7 +3173,11 @@ export class EventsService extends PuterService { compileMatch(key, { separator: KV_MATCH_SEPARATOR }); const share = await this.stores.kvShareHandle.getByHandle(handle); - if (!share || share.revokedAt !== null) + if ( + !share || + share.revokedAt !== null || + (await this.stores.kvShareHandle.predatesItsApp(share)) + ) throw unknownKvShareHandle(handle); if ( actor.user?.id !== share.granteeUserId && diff --git a/src/backend/services/kvShare/KvShareService.ts b/src/backend/services/kvShare/KvShareService.ts index c56a47f96..19a9c2ff1 100644 --- a/src/backend/services/kvShare/KvShareService.ts +++ b/src/backend/services/kvShare/KvShareService.ts @@ -17,6 +17,7 @@ * along with this program. If not, see . */ +import { randomUUID } from 'node:crypto'; import { EVENTS_KV_HANDLE_LIMIT, EVENTS_KV_HANDLES_PER_APP, @@ -56,6 +57,8 @@ import type { UserRow } from '../../stores/user/UserStore.js'; import { runWithConcurrencyLimit } from '../../util/concurrency.js'; import type { PageResult } from '../../util/pagination.js'; import { withRedisLock } from '../../util/redisLock.js'; +import { appGenerationStart, predatesApp } from '../../util/appGeneration.js'; +import { PermissionUtil } from '../permission/permissionUtil.js'; import { kvShareHandleDisabled, unknownKvShareHandle, @@ -126,6 +129,15 @@ export interface KvShareMintOptions { type Grantee = UserRow & { id: number; username: string }; +/** What a share that doesn't exist yet is made of. */ +interface NewShare { + ownerUserId: number; + appUid: string; + keyPrefix: string; + permission: string; + label: string | null; +} + interface KvShareMintResult { share?: KvShareView; error?: HttpError; @@ -321,15 +333,136 @@ const PURGE_BATCH = 500; const PURGE_PAUSE_MS = 50; const PURGE_MAX_BATCHES = 2_000; -/** Passes one app deletion takes, so a huge namespace cannot pin a node. */ -const APP_RETIRE_MAX_PASSES = 200; - -/** Items one app-data purge step deletes, and the steps one purge takes. */ +/** Items one app-data purge call deletes. */ const APP_DATA_PURGE_STEP = 5_000; -const APP_DATA_PURGE_MAX_STEPS = 2_000; -/** Deleted apps whose handles one query looks up. */ -const APP_RETIRE_UID_CHUNK = 200; +// -- Cleanup jobs ------------------------------------------------------- +// +// Retiring a deleted app's shares, and taking back an uninstalled app's shares +// and data, are recorded as reserved items before they run, so a failure or a +// restart resumes them. The record carries the job's lease, retry schedule and +// attempts, and a node claims it by compare-and-set. A job stays with the +// deployment that recorded or last ran it; another one takes it over only once +// it has gone unclaimed for a while past due. Records are kept apart per +// domain, so separate installs that share one table never run each other's. + +const jobKeyPrefix = (domain: string): string => `kvShareJob#${domain}#`; +/** How long one run works before handing its job back, and the lease on it. */ +const JOB_RUN_MS = 60_000; +const JOB_LEASE_MS = 5 * 60_000; +/** + * How often each node looks through the records, how many per look, and how + * long it keeps starting runs. + */ +const JOB_CHECK_MS = 60_000; +const JOB_CHECK_INITIAL_DELAY_MS = 60_000; +const JOB_CHECK_PAGE = 100; +const JOB_CHECK_BUDGET_MS = 2 * 60_000; +/** How long a due job waits for its own deployment before another takes it. */ +const JOB_ADOPT_AFTER_MS = 30 * 60_000; +/** Wait after a failed run: doubling from a minute, at most six hours. */ +const JOB_RETRY_BASE_MS = 60_000; +const JOB_RETRY_MAX_MS = 6 * 60 * 60_000; +/** Failed runs before a job stops retrying, and how long its record stays. */ +const JOB_MAX_ATTEMPTS = 10; +const JOB_FAILED_RETENTION_SECONDS = 30 * 24 * 60 * 60; +/** Deleted apps whose retire one call runs at once. */ +const JOB_START_CONCURRENCY = 4; + +/** + * Work recorded before it runs. `cutoff` (unix seconds) bounds it to handles + * made by then: a later one belongs to an app registered again under the same + * uid, or to a share made after the app was authorized again. + */ +type KvShareJob = + | { kind: 'app'; appUid: string; cutoff: number } + | { + kind: 'uninstall'; + userId: number; + userUuid: string; + appUid: string; + cutoff: number; + }; + +/** A job's record: the job, and the state every node claims it by. */ +interface KvShareJobRecord { + job: KvShareJob; + failed: boolean; + attempts: number; + /** Unix ms before which it isn't due. */ + runAfter: number; + /** The deployment that recorded or last claimed it. */ + home: string | null; + /** Unix ms it was last recorded or claimed. */ + touched: number; + /** Unix ms its current lease runs out; 0 when nobody holds it. */ + leaseUntil: number; +} + +interface KvShareJobRun { + done: boolean; + retired: number; +} + +const JOB_NOT_RUN: KvShareJobRun = { done: false, retired: 0 }; + +const jobKey = (prefix: string, job: KvShareJob): string => + job.kind === 'app' + ? `${prefix}app#${job.appUid}` + : `${prefix}uninstall#${job.userId}#${job.appUid}`; + +const finiteOr0 = (value: unknown): number => { + const n = Number(value); + return Number.isFinite(n) ? n : 0; +}; + +/** A recorded job, or `null` for an item that isn't one. */ +const parseJobRecord = ( + prefix: string, + item: Record, +): KvShareJobRecord | null => { + const appUid = item.appUid; + const cutoff = Number(item.cutoff); + if (typeof appUid !== 'string' || !Number.isFinite(cutoff)) return null; + let job: KvShareJob; + if (item.kind === 'app') job = { kind: 'app', appUid, cutoff }; + else if (item.kind === 'uninstall') { + const userId = Number(item.userId); + if (!Number.isInteger(userId) || typeof item.userUuid !== 'string') + return null; + job = { + kind: 'uninstall', + userId, + userUuid: item.userUuid, + appUid, + cutoff, + }; + } else return null; + if (item.key !== jobKey(prefix, job)) return null; + return { + job, + failed: item.state === 'failed', + attempts: finiteOr0(item.attempts), + runAfter: finiteOr0(item.runAfter), + home: typeof item.home === 'string' ? item.home : null, + touched: finiteOr0(item.touched), + leaseUntil: finiteOr0(item.leaseUntil), + }; +}; + +/** + * What a claim requires of the record: not given up on, unleased, due, and this + * deployment's or left unclaimed past due for long enough to adopt. + */ +const JOB_CLAIM_CONDITION = [ + 'attribute_exists(#c)', + '#st <> :failed', + '#lu < :now', + '#ra <= :now', + '(#home = :home OR (#tc < :adopt AND #ra < :adopt))', +].join(' AND '); + +const nowSeconds = (): number => Math.floor(Date.now() / 1000); const ownerView = ( row: KvShareHandle, @@ -347,11 +480,15 @@ const ownerView = ( }); export class KvShareService extends PuterService { - #purgeTimers: ReturnType[] = []; + #timers: ReturnType[] = []; #stopping = false; + #resuming = false; + /** Where this node's next look through the job records starts. */ + #jobCursor: string | undefined; override onServerStart(): void { this.#armPurge(); + this.#armJobs(); // A check at one level is answered by a grant at any level above it. this.services.permission.registerExploder(kvShareLevelsExploder()); @@ -365,15 +502,15 @@ export class KvShareService extends PuterService { // under the app-under-user generation, which removing the app bumps. this.services.permission.registerImplicator( kvShareAppDelegateImplicator({ - userHolds: (actor, permission) => - this.services.permission.check(actor, permission), + userHolds: async (actor, permission) => + (await this.services.permission.check(actor, permission)) && + (await this.#grantForCurrentApp(actor, permission)), appAuthorized: (actor) => this.#appAuthorized(actor), }), ); - // An origin-derived app uid comes back verbatim when the app is - // registered again, so a handle must not outlive its app and reattach - // to whoever controls the origin next. + // A handle must not outlive its app: an origin-derived uid comes back + // verbatim when the app is registered again. this.clients.event.on('app.changed', async (_key, data) => { const { app_uid: appUid, action } = (data ?? {}) as { app_uid?: string; @@ -383,21 +520,22 @@ export class KvShareService extends PuterService { try { await this.retireForDeletedApps([appUid]); } catch (err) { - console.warn('[kvShare] deleted-app retire failed', err); + console.warn('[kvShare] retire for a deleted app failed', err); } }); // Removing an app with its data takes back what the user shared out of - // it first, so no holder reads the namespace while it empties. + // it, then empties the namespace, in that order so no holder reads it + // while it empties. Both run in the background, off the request. this.clients.event.on('app.uninstalled', async (_key, data) => { if (!data?.deleteData || data.appUid === KV_GLOBAL_APP_KEY) return; - try { - await this.#retireOwnedInApp(data.userId, data.appUid); - } catch (err) { - console.warn('[kvShare] uninstall retire failed', err); - return; - } - void this.#purgeAppData(data.userId, data.userUuid, data.appUid); + await this.#startJob({ + kind: 'uninstall', + userId: data.userId, + userUuid: data.userUuid, + appUid: data.appUid, + cutoff: nowSeconds(), + }); }); // A developer's apps go with their account, without `app.changed`. @@ -406,15 +544,15 @@ export class KvShareService extends PuterService { try { await this.retireForDeletedApps(data.appUids); } catch (err) { - console.warn('[kvShare] deleted-account retire failed', err); + console.warn('[kvShare] retire for deleted apps failed', err); } }); } override onServerPrepareShutdown(): void { this.#stopping = true; - for (const timer of this.#purgeTimers) clearTimeout(timer); - this.#purgeTimers = []; + for (const timer of this.#timers) clearTimeout(timer); + this.#timers = []; } // -- Retention purge --------------------------------------------- @@ -468,44 +606,396 @@ export class KvShareService extends PuterService { } /** - * Empty one app's namespace for a user in bounded steps, pausing between - * them, until it is empty or the server stops. Stops if the user authorizes - * the app again, so data written after that is kept. A failure is logged, - * not retried. + * Empty an uninstalled app's namespace in steps until it's empty, the user + * authorizes the app again, or the deadline passes. The authorization is + * checked before each page is deleted, so what the app writes once it's + * authorized again is kept. Returns whether the purge is finished. */ async #purgeAppData( - userId: number, - userUuid: string, - appUid: string, - ): Promise { - const namespace = kvNamespace(userUuid, appUid); - try { - const appId = (await this.stores.app.getByUid(appUid))?.id; - for (let step = 0; step < APP_DATA_PURGE_MAX_STEPS; step++) { - if (this.#stopping) return; - if ( - typeof appId === 'number' && - ( - await this.stores.permission.listUserAppPermsFromPrimary( - userId, - appId, - ) - ).length > 0 + job: Extract, + deadline: number, + ): Promise { + const appId = (await this.stores.app.getByUid(job.appUid))?.id; + const stillWanted = async () => + typeof appId !== 'number' || + ( + await this.stores.permission.listUserAppPermsFromPrimary( + job.userId, + appId, ) - return; - const { done } = await this.stores.kv.purgeNamespace( - namespace, - { maxItems: APP_DATA_PURGE_STEP }, + ).length === 0; + if (!(await stillWanted())) return true; + + const namespace = kvNamespace(job.userUuid, job.appUid); + while (!this.#stopping && Date.now() < deadline) { + const { done, stopped } = await this.stores.kv.purgeNamespace( + namespace, + { maxItems: APP_DATA_PURGE_STEP, stillWanted }, + ); + if (done || stopped) return true; + await new Promise((resolve) => setTimeout(resolve, PURGE_PAUSE_MS)); + } + return false; + } + + // -- Cleanup jobs ------------------------------------------------- + + /** What this deployment calls itself, which is what a job's home names. */ + get #home(): string { + return this.services.broadcast.regionId; + } + + get #jobPrefix(): string { + return jobKeyPrefix(this.config.domain ?? ''); + } + + #jobKey(job: KvShareJob): string { + return jobKey(this.#jobPrefix, job); + } + + /** + * Look through the next page of recorded jobs and run the ones this node + * may claim. Each look starts where the last one stopped, so records that + * can't run yet never hide the ones after them. Every node calls this on a + * timer, which is how a failed, interrupted or unfinished job resumes. + * Returns how many finished. + */ + async resumeKvShareJobs(): Promise { + if (this.#resuming) return 0; + this.#resuming = true; + try { + const started = Date.now(); + const after = this.#jobCursor; + const page = await this.stores.kv.pageReservedItems( + this.#jobPrefix, + { limit: JOB_CHECK_PAGE, afterKey: after }, + ); + this.#jobCursor = page.nextKey; + let finished = 0; + for (const [index, item] of page.items.entries()) { + if ( + this.#stopping || + Date.now() - started > JOB_CHECK_BUDGET_MS + ) { + // The rest of this page is first in line next time. + this.#jobCursor = + index > 0 ? String(page.items[index - 1].key) : after; + break; + } + const record = parseJobRecord(this.#jobPrefix, item); + if (!record || !this.#claimable(record, Date.now())) continue; + try { + if ((await this.#runJob(this.#jobKey(record.job))).done) + finished++; + } catch { + // Alarmed, and backed off in its record. + } + } + return finished; + } finally { + this.#resuming = false; + } + } + + /** The claim condition, read off a record already in hand. */ + #claimable(record: KvShareJobRecord, now: number): boolean { + if (record.failed || record.leaseUntil >= now || record.runAfter > now) + return false; + const adopt = now - JOB_ADOPT_AFTER_MS; + return ( + record.home === this.#home || + (record.touched < adopt && record.runAfter < adopt) + ); + } + + /** Record a job, then run it in the background. */ + async #startJob(job: KvShareJob): Promise { + const recorded = await this.#recordJob(job); + void (async () => { + try { + if (recorded) await this.#runJob(this.#jobKey(job)); + else await this.#runUnrecorded(job); + } catch { + // Alarmed, and retried from its record if it has one. + } + })(); + } + + /** + * Record a job here. The cutoff only grows, so recording a job again never + * narrows it. Attempts and backoff carry over, and a job that gave up gets + * one more attempt; a lease someone holds is left alone. Returns `false`, + * alarmed, when the write failed. + */ + async #recordJob(job: KvShareJob): Promise { + const key = this.#jobKey(job); + const here = { home: this.#home, touched: Date.now() }; + const setIfAbsent = { attempts: 0, runAfter: 0, leaseUntil: 0 }; + try { + const recorded = await this.stores.kv.updateReservedItemIf( + key, + { + set: { ...job, ...here, state: 'pending' }, + setIfAbsent, + remove: ['ttl'], + }, + 'attribute_not_exists(#c) OR #c <= :c', + { ':c': job.cutoff }, + { '#c': 'cutoff' }, + ); + if (!recorded) + await this.stores.kv.updateReservedItemIf( + key, + { set: here }, + 'attribute_exists(#c)', + {}, + { '#c': 'cutoff' }, ); - if (done) return; - await new Promise((resolve) => - setTimeout(resolve, PURGE_PAUSE_MS), + return true; + } catch (err) { + this.clients.alarm.create( + `kv_share_job_record_failed:${key}`, + 'Failed to record a key-value share cleanup; running it unrecorded', + { job, error: err as Error }, + 'warning', + { dedup: true }, + ); + return false; + } + } + + /** + * A job whose record couldn't be written, run here to the end. Nothing + * resumes it, so a failure is alarmed. + */ + async #runUnrecorded(job: KvShareJob): Promise { + let retired = 0; + try { + for (;;) { + const run = await this.#jobStep(job, Date.now() + JOB_RUN_MS); + retired += run.retired; + if (run.done || this.#stopping) + return { done: run.done, retired }; + } + } catch (err) { + this.clients.alarm.create( + `kv_share_job_unrecorded_failed:${this.#jobKey(job)}`, + 'An unrecorded key-value share cleanup failed and will not be retried', + { job, retired, error: err as Error }, + 'warning', + { dedup: true }, + ); + throw err; + } + } + + /** + * One run of a recorded job: claimed off its record, then worked for at + * most `JOB_RUN_MS`. Finished, the record is deleted unless the job was + * recorded again meanwhile; unfinished, it's handed back due now. A failure + * is alarmed, backs the record off and rethrows. A job that isn't + * claimable, or that another node claimed first, reads as not done. + */ + async #runJob(key: string): Promise { + const lease = randomUUID(); + const now = Date.now(); + let claimed: Record | null; + try { + claimed = await this.stores.kv.updateReservedItemIf( + key, + { + set: { + leaseOwner: lease, + leaseUntil: now + JOB_LEASE_MS, + home: this.#home, + touched: now, + }, + }, + JOB_CLAIM_CONDITION, + { + ':failed': 'failed', + ':now': now, + ':home': this.#home, + ':adopt': now - JOB_ADOPT_AFTER_MS, + }, + { + '#c': 'cutoff', + '#st': 'state', + '#lu': 'leaseUntil', + '#ra': 'runAfter', + '#home': 'home', + '#tc': 'touched', + }, + ); + } catch (err) { + // The record stays, and the next check claims it. + console.warn('[kvShare] could not claim a cleanup job', err); + return JOB_NOT_RUN; + } + const record = claimed && parseJobRecord(this.#jobPrefix, claimed); + if (!record) return JOB_NOT_RUN; + + let run: KvShareJobRun; + try { + run = await this.#jobStep(record.job, now + JOB_RUN_MS); + } catch (err) { + await this.#jobFailed(key, lease, record, err); + throw err; + } + if (run.done) await this.#jobFinished(key, lease, record.job); + else { + console.log( + `[kvShare] cleanup ${key} continues after retiring ${run.retired}`, + ); + try { + await this.#releaseJob(key, lease, { runAfter: 0 }); + } catch (err) { + // The lease runs out and the next claim carries on. + console.warn( + '[kvShare] could not hand back a cleanup job', + err, ); } - console.warn('[kvShare] app data purge hit its step cap'); - } catch (err) { - console.warn('[kvShare] app data purge failed', err); } + return run; + } + + async #jobStep(job: KvShareJob, deadline: number): Promise { + // A share made after the cutoff isn't this job's. + const owed = (row: KvShareHandle) => row.createdAt > job.cutoff; + if (job.kind === 'app') + return this.#retireAsSystem( + () => + this.stores.kvShareHandle.listLiveForApps( + [job.appUid], + KV_HANDLE_RETIRE_BATCH, + job.cutoff, + ), + 'kv share app deleted', + deadline, + owed, + ); + // Handles the user holds in this app are left alone. + const retire = await this.#retireAsSystem( + () => + this.stores.kvShareHandle.listLiveForOwnerApp( + job.userId, + job.appUid, + KV_HANDLE_RETIRE_BATCH, + job.cutoff, + ), + 'kv share app data deleted', + deadline, + owed, + ); + if (!retire.done) return retire; + return { + done: await this.#purgeAppData(job, deadline), + retired: retire.retired, + }; + } + + /** Delete a finished job's record, or hand it back if it was recorded again. */ + async #jobFinished( + key: string, + lease: string, + job: KvShareJob, + ): Promise { + try { + const deleted = await this.stores.kv.deleteReservedItemIf( + key, + '#lo = :lease AND #c = :c', + { ':lease': lease, ':c': job.cutoff }, + { '#lo': 'leaseOwner', '#c': 'cutoff' }, + ); + if (!deleted) await this.#releaseJob(key, lease, {}); + } catch (err) { + // The record runs again and finds nothing left to do. + console.warn('[kvShare] could not clear a finished job', err); + } + } + + /** + * Count a failed run in the record and back it off, or, past + * `JOB_MAX_ATTEMPTS`, mark it failed for good and let it expire. + */ + async #jobFailed( + key: string, + lease: string, + record: KvShareJobRecord, + err: unknown, + ): Promise { + const attempts = record.attempts + 1; + const gaveUp = attempts >= JOB_MAX_ATTEMPTS; + try { + await this.#releaseJob( + key, + lease, + gaveUp + ? { + attempts, + state: 'failed', + // Due at once if it's recorded again. + runAfter: 0, + ttl: nowSeconds() + JOB_FAILED_RETENTION_SECONDS, + } + : { + attempts, + runAfter: + Date.now() + + Math.min( + JOB_RETRY_BASE_MS * 2 ** (attempts - 1), + JOB_RETRY_MAX_MS, + ), + }, + ); + } catch { + // The lease runs out and the next claim retries it. + } + this.clients.alarm.create( + gaveUp + ? `kv_share_job_gave_up:${key}` + : `kv_share_job_failed:${key}`, + gaveUp + ? 'A key-value share cleanup kept failing and stopped retrying' + : 'A key-value share cleanup failed and will be retried', + { job: record.job, attempts, error: err as Error }, + 'warning', + { dedup: true }, + ); + } + + /** Give up this node's lease on a job, writing `set` with it. */ + async #releaseJob( + key: string, + lease: string, + set: Record, + ): Promise { + await this.stores.kv.updateReservedItemIf( + key, + { set: { ...set, leaseUntil: 0, touched: Date.now() } }, + '#lo = :lease', + { ':lease': lease }, + { '#lo': 'leaseOwner' }, + ); + } + + #armJobs(): void { + const run = async () => { + try { + await this.resumeKvShareJobs(); + } catch (err) { + console.warn('[kvShare] job check failed', err); + } + }; + const kick = setTimeout( + () => void run(), + JOB_CHECK_INITIAL_DELAY_MS * (0.5 + Math.random()), + ); + kick.unref?.(); + const timer = setInterval(() => void run(), JOB_CHECK_MS); + timer.unref?.(); + this.#timers.push(kick, timer); } #armPurge(): void { @@ -526,7 +1016,7 @@ export class KvShareService extends PuterService { kick.unref?.(); const timer = setInterval(() => void run(), PURGE_CHECK_MS); timer.unref?.(); - this.#purgeTimers = [kick, timer]; + this.#timers.push(kick, timer); } /** Whether the level a mint asks for is available on this server. */ @@ -799,18 +1289,60 @@ export class KvShareService extends PuterService { throw handleQuotaReached(EVENTS_KV_HANDLES_PER_HOLDER, 'holder'); } - /** One new share: the grant, then its handle, then the holder's notice. */ + /** + * One new share: the grant, then its handle, then the holder's notice. One + * mint per region at a time, so racing mints of it make one handle: the + * ones that wait their turn get it back as it stands. + */ async #grantAndMint( actor: Actor, grantee: Grantee, - share: { - ownerUserId: number; - appUid: string; - keyPrefix: string; - permission: string; - label: string | null; - }, + share: NewShare, ): Promise { + const region = { + ownerUserId: share.ownerUserId, + granteeUserId: grantee.id, + appUid: share.appUid, + keyPrefix: share.keyPrefix, + }; + const mintUnlessLive = async () => { + const existing = await this.stores.kvShareHandle.findLive(region); + if (existing) return { row: existing, minted: false }; + return { + row: await this.#grantAndInsert(actor, grantee, share), + minted: true, + }; + }; + // A mint that waits out the lock looks again and goes ahead: the + // holder is far likelier slow than doing anything but this same mint. + const { row, minted } = await withRedisLock( + this.clients.redis, + `kvShare:mint:${region.ownerUserId}:${region.granteeUserId}:` + + `${region.appUid}:${region.keyPrefix}`, + mintUnlessLive, + { + ttlMs: SHARE_CHANGE_LOCK_MS, + attempts: SHARE_CHANGE_LOCK_ATTEMPTS, + retryMs: SHARE_CHANGE_LOCK_RETRY_MS, + onUnavailable: 'run', + onBusy: () => mintUnlessLive(), + }, + ); + if (minted) await this.#notifyReceived(row); + return row; + } + + /** {@link #grantAndMint} under its lock, with no live handle on the region. */ + async #grantAndInsert( + actor: Actor, + grantee: Grantee, + share: NewShare, + ): Promise { + const reused = await this.#retireOlderThanApp( + share.ownerUserId, + grantee.id, + share.appUid, + ); // The user behind the app, so the authority checked is theirs. Which // app acted goes on the audit row, not in the grant's authority. const authority = userRelatedActor(actor); @@ -836,13 +1368,15 @@ export class KvShareService extends PuterService { stale, meta, ); - await this.services.permission.grantUserUserPermission( - authority, - grantee.username, - share.permission, - {}, - meta, - ); + const grant = () => + this.services.permission.grantUserUserPermission( + authority, + grantee.username, + share.permission, + {}, + meta, + ); + await grant(); const row = await this.stores.kvShareHandle.mint({ ownerUserId: share.ownerUserId, @@ -852,7 +1386,9 @@ export class KvShareService extends PuterService { permission: share.permission, label: share.label, }); - await this.#notifyReceived(row); + // An older handle retired elsewhere can withdraw this region before + // the row stands, where that retire can't see it to restore the grant. + if (reused) await grant(); return row; } @@ -1123,6 +1659,88 @@ export class KvShareService extends PuterService { throw unknownKvShareHandle(named); } + /** + * `appUid`'s app when its uid was held by a deleted app before, so handles + * in its namespace can be older than it; `null` otherwise, which spares the + * caller looking at handles at all. + */ + async #reusedUidApp( + appUid: string, + ): Promise<{ created_epoch?: unknown } | null> { + if (appUid === KV_GLOBAL_APP_KEY) return null; + const app = await this.stores.app.getByUid(appUid); + if (appGenerationStart(app) === null) return null; + const deleted = await this.stores.app.getDeletedOwnersByUids([appUid]); + return deleted.has(appUid) ? app : null; + } + + /** + * Whether a share grant the user holds was made for the app its namespace + * names now, not an earlier app under the same uid: a live handle of theirs + * no older than the app covers the region at the level asked. + */ + async #grantForCurrentApp( + user: Actor, + permission: string, + ): Promise { + const [, ownerUuid, appUid] = PermissionUtil.split(permission); + if (!ownerUuid || !appUid) return true; + const app = await this.#reusedUidApp(appUid); + if (!app) return true; + + const owner = await this.stores.user.getByUuid(ownerUuid); + const holderId = user.user?.id; + if (owner?.id === undefined || typeof holderId !== 'number') + return false; + const held = await this.stores.kvShareHandle.listLiveForOwnerAndGrantee( + owner.id, + holderId, + ); + return held.some( + (row) => + row.appUid === appUid && + !predatesApp(row.createdAt, app) && + kvShareGrantCovers(row.permission, permission) && + levelRank(row.permission) >= levelRank(permission), + ); + } + + /** + * Retire, grants and all, the live handles this owner gave this holder in + * `appUid`'s namespace for an earlier app under the uid. Done before a new + * share between them lands, so no withdrawal of the old ones can take the + * new grant with it, and no old grant outlives them inside its region. + * Returns whether the uid was reused. + */ + async #retireOlderThanApp( + ownerUserId: number, + granteeUserId: number, + appUid: string, + ): Promise { + const app = await this.#reusedUidApp(appUid); + if (!app) return false; + let older: KvShareHandle[] = ( + await this.stores.kvShareHandle.listLiveForOwnerAndGrantee( + ownerUserId, + granteeUserId, + ) + ).filter( + (row) => row.appUid === appUid && predatesApp(row.createdAt, app), + ); + if (older.length === 0) return true; + await this.#retireAsSystem( + async () => { + const batch = older; + older = []; + return batch; + }, + 'kv share app uid reused', + Infinity, + (row) => !predatesApp(row.createdAt, app), + ); + return true; + } + /** Whether the actor's user still has its app authorized. */ async #appAuthorized(actor: Actor): Promise { const userId = actor.user?.id; @@ -1294,74 +1912,136 @@ export class KvShareService extends PuterService { } /** - * Retire every live handle in deleted apps' namespaces as the platform's, - * in bounded passes. + * Retire every live handle in deleted apps' namespaces as the platform's. + * Each app's retire is recorded before it runs, and what one run doesn't + * finish carries on in the background. Resolves with the handles retired + * here; rejects, after alarming, when a run failed. */ async retireForDeletedApps(appUids: readonly string[]): Promise { + const cutoff = nowSeconds(); const uids = [...new Set(appUids)].filter( (uid) => uid !== KV_GLOBAL_APP_KEY, ); let retired = 0; - for (let at = 0; at < uids.length; at += APP_RETIRE_UID_CHUNK) { - const chunk = uids.slice(at, at + APP_RETIRE_UID_CHUNK); - retired += await this.#retireAsSystem( - () => this.stores.kvShareHandle.listLiveForApps(chunk), - 'kv share app deleted', - ); - } + let failure: unknown; + await runWithConcurrencyLimit( + uids, + JOB_START_CONCURRENCY, + async (appUid) => { + try { + retired += (await this.#retireApp(appUid, cutoff)).retired; + } catch (err) { + failure ??= err; + } + }, + ); + if (failure !== undefined) throw failure; return retired; } /** - * Retire the live handles one user holds out in one app's namespace, as the - * platform's. Handles shared with the user are left alone. + * Retire, in the background, the live handles an earlier app under `app`'s + * uid left in its namespace. They already read as gone, being older than + * the app; this withdraws their grants and ends what was made through them. + * Never throws. */ - async #retireOwnedInApp( - ownerUserId: number, - appUid: string, - ): Promise { - return this.#retireAsSystem( - () => - this.stores.kvShareHandle.listLiveForOwnerApp( - ownerUserId, - appUid, - ), - 'kv share app data deleted', - ); + retireSharesOlderThanApp(app: { uid: string; created_epoch?: unknown }) { + const start = appGenerationStart(app); + if (start === null || app.uid === KV_GLOBAL_APP_KEY) return; + void (async () => { + try { + await this.#retireApp(app.uid, start - 1); + } catch (err) { + console.warn('[kvShare] retire of older shares failed', err); + } + })(); } /** - * Retire what `nextBatch` lists until a batch comes back short. Grants go - * before handles, so a retry after a failure still finds them, and a later - * share of the same region can't inherit them. + * One app's retire: nothing to record when it has no live handles. A check + * that fails is recorded anyway, so the job retries. + */ + async #retireApp(appUid: string, cutoff: number): Promise { + const job: KvShareJob = { kind: 'app', appUid, cutoff }; + try { + const live = await this.stores.kvShareHandle.listLiveForApps( + [appUid], + 1, + cutoff, + ); + if (live.length === 0) return { done: true, retired: 0 }; + } catch (err) { + console.warn('[kvShare] could not check for live shares', err); + } + if (!(await this.#recordJob(job))) return this.#runUnrecorded(job); + return this.#runJob(this.#jobKey(job)); + } + + /** + * Retire every live handle a user holds, before their account's rows + * cascade away, and tell each owner: what they subscribed to through it + * ends. The grants go with the account. + */ + async retireForDeletedHolder(userId: number): Promise { + let retired = 0; + let afterId = 0; + for (;;) { + const batch = await this.stores.kvShareHandle.listLiveHeldBy( + userId, + afterId, + ); + if (batch.length === 0) return retired; + await this.stores.kvShareHandle.retireMany( + batch.map((row) => row.handle), + 'system', + ); + await this.#announceRetired(batch, userId); + retired += batch.length; + if (batch.length < KV_HANDLE_RETIRE_BATCH) return retired; + afterId = batch[batch.length - 1].id; + } + } + + /** + * Retire what `nextBatch` lists until a batch comes back short or the + * deadline passes. Grants go before handles, so a retry after a failure + * still finds them, and a later share of the same region can't inherit + * them. `owed` names the live handles this retire isn't for. */ async #retireAsSystem( nextBatch: () => Promise, reason: string, - ): Promise { + deadline: number, + owed: (row: KvShareHandle) => boolean, + ): Promise { let retired = 0; - for (let pass = 0; pass < APP_RETIRE_MAX_PASSES; pass++) { + for (;;) { const batch = await nextBatch(); - await this.#withdrawRetiredGrants(batch, reason); + await this.#withdrawRetiredGrants(batch, reason, owed); await this.stores.kvShareHandle.retireMany( batch.map((row) => row.handle), 'system', ); await this.#announceRetired(batch); retired += batch.length; - if (batch.length < KV_HANDLE_RETIRE_BATCH) break; + if (batch.length < KV_HANDLE_RETIRE_BATCH) + return { done: true, retired }; + if (this.#stopping || Date.now() >= deadline) + return { done: false, retired }; } - return retired; } /** * Each handle's region, at every level and beneath, withdrawn in one revoke - * per owner, holder and app. + * per owner, holder and app. A live handle `owed` its grant inside one of + * those regions, such as a share made meanwhile, gets it back. */ async #withdrawRetiredGrants( rows: readonly KvShareHandle[], reason: string, + owed: (row: KvShareHandle) => boolean, ): Promise { + const retiring = new Set(rows.map((row) => row.handle)); const groups = new Map(); for (const row of rows) { const key = `${row.ownerUserId}:${row.granteeUserId}:${row.appUid}`; @@ -1381,14 +2061,38 @@ export class KvShareService extends PuterService { const holder = users.get(group[0].granteeUserId); // A deleted account's grants went with it. if (!owner?.uuid || !holder?.username) return; + const authority = makeActor({ user: owner }); await this.services.permission.revokeUserUserPermissionSubtree( - makeActor({ user: owner }), + authority, holder.username, group.flatMap((row) => kvShareGrantAtEveryLevel(row.permission), ), { reason, appUid: group[0].appUid }, ); + const live = + await this.stores.kvShareHandle.listLiveForOwnerAndGrantee( + group[0].ownerUserId, + group[0].granteeUserId, + ); + for (const row of live) { + if ( + row.appUid !== group[0].appUid || + retiring.has(row.handle) || + !owed(row) || + !group.some((old) => + kvShareGrantCovers(old.permission, row.permission), + ) + ) + continue; + await this.services.permission.grantUserUserPermission( + authority, + holder.username, + row.permission, + {}, + { reason: 'kv share handle', appUid: row.appUid }, + ); + } }, ); } @@ -1396,16 +2100,22 @@ export class KvShareService extends PuterService { /** * Tell listeners which handles stopped being addresses, one event per user * who could have subscribed through them (owner and holder alike), and wait - * for them: what was made through a handle ends with it. + * for them: what was made through a handle ends with it. `except` is a user + * whose account is going, and its subscriptions with it. */ - async #announceRetired(rows: readonly KvShareHandle[]): Promise { + async #announceRetired( + rows: readonly KvShareHandle[], + except?: number, + ): Promise { const byUser = new Map>(); for (const row of rows) - for (const userId of [row.granteeUserId, row.ownerUserId]) + for (const userId of [row.granteeUserId, row.ownerUserId]) { + if (userId === except) continue; byUser.set( userId, (byUser.get(userId) ?? new Set()).add(row.handle), ); + } await runWithConcurrencyLimit( [...byUser], RETIRE_SETTLE_CONCURRENCY, diff --git a/src/backend/services/kvShare/kvShareLevels.integration.test.ts b/src/backend/services/kvShare/kvShareLevels.integration.test.ts index e151e44fb..51ef3ce14 100644 --- a/src/backend/services/kvShare/kvShareLevels.integration.test.ts +++ b/src/backend/services/kvShare/kvShareLevels.integration.test.ts @@ -1123,7 +1123,16 @@ describe('an app deletion that fails partway', () => { } expect((await handles().getByHandle(handle))?.revokedAt).toBeNull(); - await deleted(); + // The next check after its backoff retries it. + const now = Date.now.bind(Date); + const clock = vi + .spyOn(Date, 'now') + .mockImplementation(() => now() + 2 * 60_000); + try { + await shares().resumeKvShareJobs(); + } finally { + clock.mockRestore(); + } expect(await grantRows(guest, prefix, app.uid)).toEqual([]); expect(await handles().getByHandle(handle)).toMatchObject({ revokedBy: 'system', diff --git a/src/backend/services/kvShare/kvShareLifecycle.integration.test.ts b/src/backend/services/kvShare/kvShareLifecycle.integration.test.ts new file mode 100644 index 000000000..60892d9a4 --- /dev/null +++ b/src/backend/services/kvShare/kvShareLifecycle.integration.test.ts @@ -0,0 +1,833 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +/** + * What happens to shares when what they hang off goes away: an app deleted and + * registered again under the same uid, a deletion whose retire fails or runs + * long, a holder's account deleted, and racing shares of one region. + */ + +import { v4 as uuidv4 } from 'uuid'; +import { + afterAll, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from 'vitest'; +import { EVENTS_KV_HANDLE_LIMIT } from '../../controllers/events/limits.js'; +import { makeActor, type Actor } from '../../core/actor.js'; +import { runWithContext } from '../../core/context.js'; +import { KV_HANDLE_RETIRE_BATCH } from '../../stores/events/KvShareHandleStore.js'; +import { + createTestUser, + setupPuterTestEnv, + type PuterTestEnv, +} from '../../testUtil.js'; +import type { IConfig } from '../../types.js'; +import { kvSharePermission, type KvShareAccess } from './kvShares.js'; +import { seedKvShareHandles } from './kvShareTestUtil.js'; + +const BOOT_TIMEOUT_MS = 120_000; + +interface TestUser { + actor: Actor; + username: string; + id: number; + uuid: string; +} + +let env: PuterTestEnv; +let owner: TestUser; +let guest: TestUser; + +const shares = () => env.server.services.kvShare; +const handles = () => env.server.stores.kvShareHandle; + +const userFor = async (username: string): Promise => { + const row = await env.server.stores.user.getByUsername(username); + return { + actor: makeActor({ user: row as never }), + username, + id: row!.id, + uuid: row!.uuid as string, + }; +}; + +const newUser = async (label: string): Promise => { + const username = `${label}-${uuidv4().slice(0, 8)}`; + await createTestUser(env.server, { username, password: 'pw-test-1234' }); + return userFor(username); +}; + +const region = (label = 'r') => `${label}-${uuidv4().slice(0, 8)}:`; + +const share = ( + prefix: string, + access: KvShareAccess, + opts: { holder?: TestUser; appUid?: string } = {}, +) => + shares().mintKvHandle( + owner.actor, + { + prefix, + recipient: { username: (opts.holder ?? guest).username }, + access, + ...(opts.appUid ? { appUid: opts.appUid } : {}), + }, + { defaultAccess: 'read' }, + ); + +const makeApp = async () => { + const name = `kv-life-${uuidv4().slice(0, 8)}`; + return env.server.stores.app.create( + { + name, + title: 'Lifecycle App', + index_url: `https://${name}.example.test/index.html`, + }, + { ownerUserId: owner.id }, + ); +}; + +/** An app deleted the way the app driver does it: the row, then the event. */ +const deleteApp = async (app: { id: number; uid: string }) => { + await env.server.stores.app.delete(app.id); + await env.server.clients.event.emitAndWait( + 'app.changed', + { app_uid: app.uid, app: null, old_app: app, action: 'deleted' }, + {}, + ); +}; + +const liveInApp = async (appUid: string) => + (await handles().listLiveForApps([appUid], 5_000)).length; + +const guestGet = (handle: string, key: string, actor = guest.actor) => + runWithContext({ actor }, () => + env.server.drivers.kvStore.get({ key, optConfig: { handle } }), + ); + +const ownerWrites = (key: string, value: unknown, appUid: string) => + runWithContext({ actor: owner.actor }, () => + env.server.drivers.kvStore.set({ + key, + value, + optConfig: { appUuid: appUid }, + }), + ); + +/** Make a handle look minted `seconds` ago, past the clock-skew allowance. */ +const age = (handle: string, seconds = 2 * 60 * 60) => + env.server.clients.db.write( + 'UPDATE `kv_share_handles` SET `created_at` = `created_at` - ? WHERE `handle` = ?', + [seconds, handle], + ); + +/** The job record for one deleted app, as every node of this domain reads it. */ +const appJobKey = (appUid: string, domain = new URL(env.origin).hostname) => + `kvShareJob#${domain}#app#${appUid}`; + +const suspendedReason = async (subId: string) => { + const [row] = await env.server.clients.db.pread( + 'SELECT `suspended_reason` FROM `event_subscriptions` WHERE `sub_id` = ?', + [subId], + ); + return (row as { suspended_reason: unknown } | undefined)?.suspended_reason; +}; + +/** Run `fn` with the clock `ms` ahead, past any retry wait shorter than that. */ +const later = async (ms: number, fn: () => Promise): Promise => { + const now = Date.now.bind(Date); + const clock = vi.spyOn(Date, 'now').mockImplementation(() => now() + ms); + try { + return await fn(); + } finally { + clock.mockRestore(); + } +}; + +/** Check for recorded jobs until `check` passes. */ +const resumeUntil = (check: () => Promise) => + vi.waitFor( + async () => { + await shares().resumeKvShareJobs(); + await check(); + }, + { timeout: 4_000, interval: 100 }, + ); + +beforeAll(async () => { + env = await setupPuterTestEnv({ + events: { enabled: true, kvHandles: true }, + kv: { shareHandles: true }, + unlimitedMetering: true, + } as IConfig); + owner = await userFor(env.users.user.username); + guest = await userFor(env.users.other.username); +}, BOOT_TIMEOUT_MS); + +afterAll(async () => { + await env?.shutdown(); +}); + +beforeEach(async () => { + for (const user of [owner, guest]) + await env.server.clients.redis.del( + `rate:${EVENTS_KV_HANDLE_LIMIT.scope}:${user.id}`, + ); +}); + +describe('an origin app registered again under a deleted app`s uid', () => { + const call = (body: unknown) => + fetch(new URL('/auth/get-user-app-token', env.apiOrigin), { + method: 'POST', + headers: { + authorization: `Bearer ${env.users.user.token}`, + 'content-type': 'application/json', + }, + body: JSON.stringify(body), + }); + + const originFor = (label: string) => + `https://${label}-${uuidv4().slice(0, 8)}.example.com`; + + /** + * An origin app with an old share in it, deleted without its retire + * running, so the handle and its grant are still live. + */ + const deletedWithShare = async () => { + const origin = originFor('kv-reuse'); + const appUid = await env.server.services.auth.appUidFromOrigin(origin); + const app = await env.server.stores.app.createFromOrigin( + appUid, + origin, + ); + const prefix = region('reuse'); + const { handle } = await share(prefix, 'read', { appUid }); + await age(handle); + await env.server.stores.app.delete(app.id); + return { origin, appUid, prefix, handle }; + }; + + it('refuses a handle older than the app, through the driver and the delegate, with nothing retired', async () => { + const { origin, appUid, prefix, handle } = await deletedWithShare(); + // Registered again the way any bootstrap does, with no retire. + const app = await env.server.stores.app.createFromOrigin( + appUid, + origin, + ); + await env.server.services.permission.grantUserAppPermission( + guest.actor, + appUid, + 'flag:app-is-authenticated', + ); + const guestInApp = makeActor({ + user: guest.actor.user as never, + app: { uid: appUid, id: app.id }, + }); + + for (const actor of [guest.actor, guestInApp]) + await expect(guestGet(handle, 'k', actor)).rejects.toMatchObject({ + legacyCode: 'kv_handle_not_found', + }); + expect( + await env.server.services.permission.check( + guestInApp, + kvSharePermission(owner.uuid, appUid, prefix, 'read'), + ), + ).toBe(false); + expect((await handles().getByHandle(handle))?.revokedAt).toBeNull(); + + // A share made for the new app works as usual. + const fresh = await share(region('fresh'), 'read', { appUid }); + expect(await guestGet(fresh.handle, 'k', guestInApp)).toBeNull(); + expect( + await env.server.services.permission.check( + guestInApp, + kvSharePermission(owner.uuid, appUid, fresh.prefix, 'read'), + ), + ).toBe(true); + }); + + it('makes a new share when the owner shares an old region again', async () => { + const { origin, appUid, prefix, handle } = await deletedWithShare(); + const app = await env.server.stores.app.createFromOrigin( + appUid, + origin, + ); + + const again = await share(prefix, 'read', { appUid }); + + expect(again.handle).not.toBe(handle); + expect(await handles().getByHandle(handle)).toMatchObject({ + revokedBy: 'system', + }); + await ownerWrites(`${prefix}k`, 'v', appUid); + expect(await guestGet(again.handle, 'k')).toBe('v'); + + // The cleanup a launch starts leaves the new share's grant alone. + shares().retireSharesOlderThanApp(app); + await new Promise((resolve) => setTimeout(resolve, 300)); + expect(await guestGet(again.handle, 'k')).toBe('v'); + }); + + it('leaves the old shares out of both sides` listings', async () => { + const { origin, appUid, handle } = await deletedWithShare(); + const app = await env.server.stores.app.createFromOrigin( + appUid, + origin, + ); + const guestInApp = makeActor({ + user: guest.actor.user as never, + app: { uid: appUid, id: app.id }, + }); + + for (const actor of [guest.actor, guestInApp]) { + const held = await shares().listKvSharesWithMe(actor, { + appUid, + includeTotal: true, + }); + expect(held.items).toEqual([]); + expect(held.total).toBe(0); + } + for (const state of ['all', 'live'] as const) { + const mine = await shares().listKvHandles(owner.actor, { + appUid, + state, + includeTotal: true, + }); + expect(mine.items).toEqual([]); + expect(mine.total).toBe(0); + } + + // Once the cleanup retires it, the owner sees when it stopped. + shares().retireSharesOlderThanApp(app); + await vi.waitFor(async () => + expect( + (await shares().listKvHandles(owner.actor, { appUid })).items, + ).toEqual([ + expect.objectContaining({ handle, revokedBy: 'system' }), + ]), + ); + }); + + it('launches without waiting on the retire, which cleans up in the background', async () => { + const { origin, appUid, handle } = await deletedWithShare(); + + let release!: () => void; + const held = new Promise((resolve) => { + release = resolve; + }); + const store = handles(); + const retireMany = store.retireMany.bind(store); + const stalled = vi + .spyOn(store, 'retireMany') + .mockImplementation(async (...args) => { + await held; + return retireMany(...args); + }); + try { + const res = await call({ origin }); + expect(res.status).toBe(200); + expect(((await res.json()) as { app_uid: string }).app_uid).toBe( + appUid, + ); + await expect(guestGet(handle, 'k')).rejects.toMatchObject({ + legacyCode: 'kv_handle_not_found', + }); + } finally { + release(); + stalled.mockRestore(); + } + await vi.waitFor(async () => + expect(await handles().getByHandle(handle)).toMatchObject({ + revokedBy: 'system', + }), + ); + }); + + it('launches when the retire fails', async () => { + const { origin, handle } = await deletedWithShare(); + + const failing = vi + .spyOn( + env.server.services.permission, + 'revokeUserUserPermissionSubtree', + ) + .mockRejectedValue(new Error('database went away')); + try { + expect((await call({ origin })).status).toBe(200); + await expect(guestGet(handle, 'k')).rejects.toMatchObject({ + legacyCode: 'kv_handle_not_found', + }); + } finally { + failing.mockRestore(); + } + }); +}); + +describe('a deleted app`s retire', () => { + it('alarms when it fails, and the next check retries it from its record', async () => { + const app = await makeApp(); + const { handle } = await share(region('retry'), 'read', { + appUid: app.uid, + }); + + const alarm = vi.spyOn(env.server.clients.alarm, 'create'); + const failing = vi + .spyOn( + env.server.services.permission, + 'revokeUserUserPermissionSubtree', + ) + .mockRejectedValueOnce(new Error('database went away')); + try { + await deleteApp(app); + expect(alarm).toHaveBeenCalledWith( + expect.stringContaining(app.uid), + expect.any(String), + expect.objectContaining({ error: expect.any(Error) }), + 'warning', + { dedup: true }, + ); + } finally { + failing.mockRestore(); + alarm.mockRestore(); + } + expect((await handles().getByHandle(handle))?.revokedAt).toBeNull(); + + // Backing off after the failure, so an immediate check leaves it. + await shares().resumeKvShareJobs(); + expect((await handles().getByHandle(handle))?.revokedAt).toBeNull(); + + await later(2 * 60_000, () => + resumeUntil(async () => + expect(await handles().getByHandle(handle)).toMatchObject({ + revokedBy: 'system', + }), + ), + ); + }); + + it('runs anyway, with an alarm, when its record can`t be written', async () => { + const app = await makeApp(); + const { handle } = await share(region('unrecorded'), 'read', { + appUid: app.uid, + }); + const alarm = vi.spyOn(env.server.clients.alarm, 'create'); + const unwritable = vi + .spyOn(env.server.stores.kv, 'updateReservedItemIf') + .mockRejectedValueOnce(new Error('throttled')); + try { + await deleteApp(app); + expect(alarm).toHaveBeenCalledWith( + `kv_share_job_record_failed:${appJobKey(app.uid)}`, + expect.any(String), + expect.objectContaining({ error: expect.any(Error) }), + 'warning', + { dedup: true }, + ); + } finally { + unwritable.mockRestore(); + alarm.mockRestore(); + } + expect(await handles().getByHandle(handle)).toMatchObject({ + revokedBy: 'system', + }); + }); + + it('keeps its backoff when it`s recorded again', async () => { + const app = await makeApp(); + const { handle } = await share(region('backoff'), 'read', { + appUid: app.uid, + }); + const failing = vi + .spyOn( + env.server.services.permission, + 'revokeUserUserPermissionSubtree', + ) + .mockRejectedValueOnce(new Error('database went away')); + try { + await deleteApp(app); + } finally { + failing.mockRestore(); + } + + // What a periodic sweep of deleted apps does. + await shares().retireForDeletedApps([app.uid]); + expect((await handles().getByHandle(handle))?.revokedAt).toBeNull(); + + await later(2 * 60_000, () => + resumeUntil(async () => + expect(await handles().getByHandle(handle)).toMatchObject({ + revokedBy: 'system', + }), + ), + ); + }); + + it('never runs a record written for another domain', async () => { + const app = await makeApp(); + const { handle } = await share(region('other-domain'), 'read', { + appUid: app.uid, + }); + await env.server.stores.app.delete(app.id); + const key = appJobKey(app.uid, 'staging.example.test'); + await env.server.stores.kv.putReservedItem(key, { + kind: 'app', + appUid: app.uid, + cutoff: Math.floor(Date.now() / 1000), + state: 'pending', + attempts: 0, + runAfter: 0, + home: env.server.services.broadcast.regionId, + touched: 0, + leaseUntil: 0, + }); + try { + await later(60 * 60_000, async () => { + for (let look = 0; look < 3; look++) + await shares().resumeKvShareJobs(); + }); + expect((await handles().getByHandle(handle))?.revokedAt).toBeNull(); + } finally { + await env.server.stores.kv.deleteReservedItemIf( + key, + 'attribute_exists(#c)', + {}, + { '#c': 'cutoff' }, + ); + } + }); + + it('gives a share made meanwhile inside a retired region its grant back', async () => { + const app = await makeApp(); + const outer = region('outer'); + const before = await share(outer, 'read', { appUid: app.uid }); + // Dated after the deletion, as one made while the retire runs would be. + const after = await share(`${outer}inner:`, 'read', { + appUid: app.uid, + }); + await env.server.clients.db.write( + 'UPDATE `kv_share_handles` SET `created_at` = `created_at` + 60 WHERE `handle` = ?', + [after.handle], + ); + + await deleteApp(app); + + expect(await handles().getByHandle(before.handle)).toMatchObject({ + revokedBy: 'system', + }); + const kept = await handles().getByHandle(after.handle); + expect(kept?.revokedAt).toBeNull(); + expect( + await env.server.services.permission.hasUserUserGrant( + guest.id, + owner.id, + [kept!.permission], + ), + ).toBe(true); + }); + + it('carries on from its record when one run runs out of time', async () => { + const app = await makeApp(); + // Spread over holders, as the per-holder cap would spread them. + const holders = await Promise.all( + [0, 1, 2].map(() => newUser('kv-seed')), + ); + const perHolder = Math.ceil((KV_HANDLE_RETIRE_BATCH + 1) / 3); + for (const holder of holders) + await seedKvShareHandles(env.server, { + ownerUserId: owner.id, + ownerUuid: owner.uuid, + granteeUserId: holder.id, + count: perHolder, + appUid: () => app.uid, + }); + const seeded = perHolder * holders.length; + + // The first batch takes the run past its deadline. + const store = handles(); + const retireMany = store.retireMany.bind(store); + const now = Date.now.bind(Date); + const clock = vi.spyOn(Date, 'now'); + const slow = vi + .spyOn(store, 'retireMany') + .mockImplementationOnce(async (...args) => { + await retireMany(...args); + clock.mockImplementation(() => now() + 2 * 60_000); + }); + try { + await deleteApp(app); + } finally { + slow.mockRestore(); + clock.mockRestore(); + } + expect(await liveInApp(app.uid)).toBe(seeded - KV_HANDLE_RETIRE_BATCH); + + await resumeUntil(async () => expect(await liveInApp(app.uid)).toBe(0)); + }); + + it('leaves alone shares made after the deletion under the same uid', async () => { + const app = await makeApp(); + const before = await share(region('old'), 'read', { appUid: app.uid }); + // Dated after the deletion, as one made by an app registered again + // under the uid would be. + const after = await share(region('new'), 'read', { appUid: app.uid }); + await env.server.clients.db.write( + 'UPDATE `kv_share_handles` SET `created_at` = `created_at` + 60 WHERE `handle` = ?', + [after.handle], + ); + + await deleteApp(app); + + expect(await handles().getByHandle(before.handle)).toMatchObject({ + revokedBy: 'system', + }); + expect( + (await handles().getByHandle(after.handle))?.revokedAt, + ).toBeNull(); + }); + + it('runs once when two runners race for it', async () => { + const app = await makeApp(); + const { handle } = await share(region('race-job'), 'read', { + appUid: app.uid, + }); + await env.server.stores.app.delete(app.id); + + const store = handles(); + const retireMany = vi.spyOn(store, 'retireMany'); + try { + await Promise.all([ + shares().retireForDeletedApps([app.uid]), + shares().retireForDeletedApps([app.uid]), + ]); + const runs = retireMany.mock.calls.filter(([named]) => + named.includes(handle), + ); + expect(runs).toHaveLength(1); + } finally { + retireMany.mockRestore(); + } + expect(await handles().getByHandle(handle)).toMatchObject({ + revokedBy: 'system', + }); + }); + + it('is left to the deployment that holds it until it goes unclaimed', async () => { + const app = await makeApp(); + const { handle } = await share(region('elsewhere'), 'read', { + appUid: app.uid, + }); + const failing = vi + .spyOn( + env.server.services.permission, + 'revokeUserUserPermissionSubtree', + ) + .mockRejectedValueOnce(new Error('database went away')); + try { + await deleteApp(app); + } finally { + failing.mockRestore(); + } + // Another deployment ran it last, a moment ago, and it's due now. + await env.server.stores.kv.updateReservedItemIf( + appJobKey(app.uid), + { set: { home: 'elsewhere', touched: Date.now(), runAfter: 0 } }, + 'attribute_exists(#c)', + {}, + { '#c': 'cutoff' }, + ); + + await later(5 * 60_000, async () => { + await shares().resumeKvShareJobs(); + await shares().resumeKvShareJobs(); + }); + expect((await handles().getByHandle(handle))?.revokedAt).toBeNull(); + + await later(31 * 60_000, () => + resumeUntil(async () => + expect(await handles().getByHandle(handle)).toMatchObject({ + revokedBy: 'system', + }), + ), + ); + }); + + it('stops retrying after repeated failures, with an alarm', async () => { + const app = await makeApp(); + const { handle } = await share(region('gave-up'), 'read', { + appUid: app.uid, + }); + const alarm = vi.spyOn(env.server.clients.alarm, 'create'); + const failing = vi + .spyOn( + env.server.services.permission, + 'revokeUserUserPermissionSubtree', + ) + .mockRejectedValue(new Error('database went away')); + const gaveUp = () => + alarm.mock.calls.some( + ([id]) => + id.startsWith('kv_share_job_gave_up:') && + id.includes(app.uid), + ); + try { + await deleteApp(app); + // Each check comes after the longest wait between retries. + for (let hours = 7; hours < 7 * 20 && !gaveUp(); hours += 7) + await later(hours * 60 * 60_000, async () => { + for (let look = 0; look < 5; look++) + await shares().resumeKvShareJobs(); + }); + expect(gaveUp()).toBe(true); + + failing.mockClear(); + await later(1_000 * 60 * 60_000, async () => { + for (let look = 0; look < 5; look++) + await shares().resumeKvShareJobs(); + }); + expect(failing).not.toHaveBeenCalled(); + } finally { + failing.mockRestore(); + alarm.mockRestore(); + } + expect((await handles().getByHandle(handle))?.revokedAt).toBeNull(); + + // A new deletion event records it again. + await deleteApp(app); + expect(await handles().getByHandle(handle)).toMatchObject({ + revokedBy: 'system', + }); + }); + + it('reaches a due job behind a page of ones that are waiting', async () => { + const waiting: string[] = []; + for (let i = 0; i < 150; i++) { + const appUid = `app-0000-wait-${String(i).padStart(4, '0')}`; + waiting.push(appJobKey(appUid)); + await env.server.stores.kv.putReservedItem(appJobKey(appUid), { + kind: 'app', + appUid, + cutoff: 0, + state: 'pending', + attempts: 1, + runAfter: Date.now() + 1_000 * 60 * 60_000, + home: env.server.services.broadcast.regionId, + touched: Date.now(), + leaseUntil: 0, + }); + } + try { + const app = await makeApp(); + const { handle } = await share(region('behind'), 'read', { + appUid: app.uid, + }); + const failing = vi + .spyOn( + env.server.services.permission, + 'revokeUserUserPermissionSubtree', + ) + .mockRejectedValueOnce(new Error('database went away')); + try { + await deleteApp(app); + } finally { + failing.mockRestore(); + } + + await later(2 * 60_000, async () => { + for (let look = 0; look < 4; look++) + await shares().resumeKvShareJobs(); + }); + expect(await handles().getByHandle(handle)).toMatchObject({ + revokedBy: 'system', + }); + } finally { + for (const key of waiting) + await env.server.stores.kv.deleteReservedItemIf( + key, + 'attribute_exists(#c)', + {}, + { '#c': 'cutoff' }, + ); + } + }); +}); + +describe('shares of one region made at the same time', () => { + it('make one handle, and every call gets it', async () => { + const prefix = region('race'); + + const minted = await Promise.all( + [0, 1, 2].map(() => share(prefix, 'read')), + ); + + expect(new Set(minted.map((one) => one.handle)).size).toBe(1); + const live = ( + await handles().listLiveForOwnerAndGrantee(owner.id, guest.id) + ).filter((row) => row.keyPrefix === prefix); + expect(live.map((row) => row.handle)).toEqual([minted[0].handle]); + }); + + it('hands a call that waited out the lock the share made meanwhile', async () => { + const prefix = region('slow'); + const store = handles(); + const mint = store.mint.bind(store); + const pause = (ms: number) => + new Promise((resolve) => setTimeout(resolve, ms)); + // The first mint inserts after the second has looked and found + // nothing, then holds the lock well past how long the second waits. + const slow = vi + .spyOn(store, 'mint') + .mockImplementationOnce(async (input) => { + await pause(300); + const row = await mint(input); + await pause(2_000); + return row; + }); + try { + const first = share(prefix, 'read'); + await pause(100); + const second = await share(prefix, 'read'); + expect(second.handle).toBe((await first).handle); + } finally { + slow.mockRestore(); + } + }); +}); + +describe('deleting an account', () => { + const subscribe = (user: TestUser, handle: string) => + env.server.services.events.subscribeDurable(user.actor, { + subject: `kv:${handle}:*`, + }); + + it('ends what the owner subscribed to through a share the account held', async () => { + const holder = await newUser('kv-holder'); + const { handle } = await share(region('held'), 'read', { holder }); + const { sub } = await subscribe(owner, handle); + + await env.server.services.userAccount.cascadeDelete(holder.id); + + expect(await suspendedReason(sub.subId)).toBe('permission_revoked'); + }); +}); diff --git a/src/backend/services/kvShare/kvShareUninstall.integration.test.ts b/src/backend/services/kvShare/kvShareUninstall.integration.test.ts index a994ad80f..faf9798ed 100644 --- a/src/backend/services/kvShare/kvShareUninstall.integration.test.ts +++ b/src/backend/services/kvShare/kvShareUninstall.integration.test.ts @@ -110,6 +110,17 @@ const revokedBy = async (handle: string) => (await env.server.stores.kvShareHandle.getByHandle(handle))?.revokedBy ?? null; +/** Run `fn` with the clock `ms` ahead, past any retry wait shorter than that. */ +const later = async (ms: number, fn: () => Promise): Promise => { + const now = Date.now.bind(Date); + const clock = vi.spyOn(Date, 'now').mockImplementation(() => now() + ms); + try { + return await fn(); + } finally { + clock.mockRestore(); + } +}; + describe('uninstalling an app', () => { it('keeps the data and the shares made out of it without deleteData', async () => { const app = await makeApp(); @@ -151,7 +162,9 @@ describe('uninstalling an app', () => { await uninstall(app.uid, true); - expect(await revokedBy(owned.handle)).toBe('system'); + await vi.waitFor(async () => + expect(await revokedBy(owned.handle)).toBe('system'), + ); expect(await revokedBy(ownedElsewhere.handle)).toBeNull(); expect(await revokedBy(held.handle)).toBeNull(); @@ -194,4 +207,149 @@ describe('uninstalling an app', () => { } expect(await keysIn(owner, app.uid)).toEqual(['after-reinstall']); }); + + it('returns before the shares are retired and the data deleted', async () => { + const app = await makeApp(); + await env.server.stores.kv.set( + { key: 'k', value: 1 }, + kvOpts(owner, app.uid), + ); + const owned = await share(owner, guest, app.uid); + + let release!: () => void; + const held = new Promise((resolve) => { + release = resolve; + }); + const store = env.server.stores.kvShareHandle; + const retireMany = store.retireMany.bind(store); + const stalled = vi + .spyOn(store, 'retireMany') + .mockImplementation(async (...args) => { + await held; + return retireMany(...args); + }); + try { + const returned = (async () => { + await uninstall(app.uid, true); + return 'returned'; + })(); + const waited = new Promise((resolve) => + setTimeout(() => resolve('waited'), 2_000), + ); + const outcome = await Promise.race([returned, waited]); + expect(outcome).toBe('returned'); + expect(await revokedBy(owned.handle)).toBeNull(); + } finally { + release(); + stalled.mockRestore(); + } + + await vi.waitFor(async () => + expect(await revokedBy(owned.handle)).toBe('system'), + ); + await vi.waitFor(async () => + expect(await keysIn(owner, app.uid)).toEqual([]), + ); + }); + + it('alarms when the cleanup fails, and the next check finishes it', async () => { + const app = await makeApp(); + await env.server.stores.kv.set( + { key: 'k', value: 1 }, + kvOpts(owner, app.uid), + ); + + const alarm = vi.spyOn(env.server.clients.alarm, 'create'); + const failing = vi + .spyOn(env.server.stores.kv, 'purgeNamespace') + .mockRejectedValueOnce(new Error('table went away')); + try { + await uninstall(app.uid, true); + await vi.waitFor(() => + expect(alarm).toHaveBeenCalledWith( + expect.stringContaining(app.uid), + expect.any(String), + expect.objectContaining({ error: expect.any(Error) }), + 'warning', + { dedup: true }, + ), + ); + } finally { + failing.mockRestore(); + alarm.mockRestore(); + } + expect(await keysIn(owner, app.uid)).toEqual(['k']); + + await later(2 * 60_000, () => + vi.waitFor( + async () => { + await env.server.services.kvShare.resumeKvShareJobs(); + expect(await keysIn(owner, app.uid)).toEqual([]); + }, + { timeout: 4_000, interval: 100 }, + ), + ); + }); + + it('keeps what the app writes once the user authorizes it again mid-purge', async () => { + const app = await makeApp(); + // More than one page of the purge. + await env.server.stores.kv.batchPut( + { + items: Array.from({ length: 1_200 }, (_, i) => ({ + key: `item-${String(i).padStart(4, '0')}`, + value: i, + })), + }, + kvOpts(owner, app.uid), + ); + + // The user opens the app again while the first page is deleted, and + // the app writes a key that sorts after every page. + const dynamo = env.server.clients.dynamo; + const batchDel = dynamo.batchDel.bind(dynamo); + let reinstalled = false; + const deleting = vi + .spyOn(dynamo, 'batchDel') + .mockImplementation(async (params) => { + if ( + !reinstalled && + params.some((p) => + String(p.key.namespace).endsWith(app.uid), + ) + ) { + reinstalled = true; + await env.server.stores.permission.upsertUserAppPerm( + owner.id, + app.id as number, + 'flag:app-is-authenticated', + {}, + ); + await env.server.stores.kv.set( + { key: 'zz-after-reinstall', value: 1 }, + kvOpts(owner, app.uid), + ); + } + return batchDel(params); + }); + try { + await uninstall(app.uid, true); + await vi.waitFor( + async () => { + expect(reinstalled).toBe(true); + const keys = await keysIn(owner, app.uid); + expect(keys).toContain('zz-after-reinstall'); + expect(keys).toHaveLength(1_200 - 1_000 + 1); + }, + { timeout: 4_000, interval: 100 }, + ); + // And stays that way: the purge stopped rather than paused. + await new Promise((resolve) => setTimeout(resolve, 500)); + expect(await keysIn(owner, app.uid)).toHaveLength( + 1_200 - 1_000 + 1, + ); + } finally { + deleting.mockRestore(); + } + }); }); diff --git a/src/backend/services/user/UserAccountService.ts b/src/backend/services/user/UserAccountService.ts index 9fbf6b40a..ad606125f 100644 --- a/src/backend/services/user/UserAccountService.ts +++ b/src/backend/services/user/UserAccountService.ts @@ -120,6 +120,15 @@ export class UserAccountService extends PuterService { ownerUserId: userId, }); + // Handles the account holds cascade with the row and nothing is told, + // so what their owners subscribed to through them would stay active + // and silent. Retiring them first ends those subscriptions. + try { + await this.services.kvShare.retireForDeletedHolder(userId); + } catch (e) { + console.warn('[cascade-delete-user] kv share retire failed:', e); + } + // Sessions FK is SET NULL, so delete explicitly to avoid dangling rows. await this.clients.db.write( 'DELETE FROM `sessions` WHERE `user_id` = ?', diff --git a/src/backend/stores/app/AppStore.js b/src/backend/stores/app/AppStore.js index 47f17e901..58fe5f4ab 100644 --- a/src/backend/stores/app/AppStore.js +++ b/src/backend/stores/app/AppStore.js @@ -1335,18 +1335,9 @@ export class AppStore extends PuterStore { return this.#normalizeRow(rows[0]); } - /** - * The row's creation time as unix seconds, computed by the database: the - * mysql driver reads a stored UTC timestamp as local time, so parsing - * `timestamp` in JS can be off by the server's UTC offset. - */ + /** The row's creation time as unix seconds. */ #createdEpochColumn(alias = '') { - const column = `${alias}\`timestamp\``; - const epoch = this.clients.db.case({ - postgres: `EXTRACT(EPOCH FROM ${column})::bigint`, - mysql: `UNIX_TIMESTAMP(${column})`, - otherwise: `CAST(strftime('%s', ${column}) AS INTEGER)`, - }); + const epoch = this.clients.db.unixSecondsSql(`${alias}\`timestamp\``); return `${epoch} AS \`created_epoch\``; } diff --git a/src/backend/stores/events/KvShareHandleStore.integration.test.ts b/src/backend/stores/events/KvShareHandleStore.integration.test.ts index 288433344..c850bb3e5 100644 --- a/src/backend/stores/events/KvShareHandleStore.integration.test.ts +++ b/src/backend/stores/events/KvShareHandleStore.integration.test.ts @@ -33,7 +33,11 @@ import { vi, } from 'vitest'; import { isHttpError } from '../../core/http/HttpError.js'; -import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js'; +import { + createTestUser, + setupPuterTestEnv, + type PuterTestEnv, +} from '../../testUtil.js'; import type { IConfig } from '../../types.js'; import { KV_SHARE_HANDLE_WIDTHS } from './columnWidths.js'; import type { MintKvShareHandleInput } from './KvShareHandleStore.js'; @@ -304,6 +308,66 @@ describe('live handles by owner and app', () => { }); }); +describe('a cutoff on the retire listings', () => { + it('leaves out handles made after it', async () => { + const appUid = 'app-cutoff'; + const old = await store().mint(input({ appUid, keyPrefix: 'old:' })); + const fresh = await store().mint( + input({ appUid, keyPrefix: 'fresh:' }), + ); + await env.server.clients.db.write( + 'UPDATE `kv_share_handles` SET `created_at` = ? WHERE `handle` = ?', + [old.createdAt - 10, old.handle], + ); + const cutoff = old.createdAt - 5; + + expect( + (await store().listLiveForApps([appUid], 10, cutoff)).map( + (row) => row.handle, + ), + ).toEqual([old.handle]); + expect( + ( + await store().listLiveForOwnerApp( + ownerUserId, + appUid, + 10, + cutoff, + ) + ).map((row) => row.handle), + ).toEqual([old.handle]); + expect(await store().listLiveForApps([appUid], 10)).toHaveLength(2); + await store().retireMany([old.handle, fresh.handle], 'owner'); + }); +}); + +describe('live handles one user holds', () => { + it('come a batch at a time after an id, live ones only', async () => { + const holderName = `kv-held-${Math.random().toString(36).slice(2, 10)}`; + await createTestUser(env.server, { + username: holderName, + password: 'pw-test-1234', + }); + const holderId = (await env.server.stores.user.getByUsername( + holderName, + ))!.id; + const held = []; + for (const keyPrefix of ['held-a:', 'held-b:', 'held-c:']) + held.push( + await store().mint( + input({ granteeUserId: holderId, keyPrefix }), + ), + ); + await store().retireMany([held[1].handle], 'owner'); + + const first = await store().listLiveHeldBy(holderId, 0, 1); + expect(first.map((row) => row.handle)).toEqual([held[0].handle]); + const rest = await store().listLiveHeldBy(holderId, first[0].id); + expect(rest.map((row) => row.handle)).toEqual([held[2].handle]); + expect(rest[0].ownerUserId).toBe(ownerUserId); + }); +}); + describe('the live cache across nodes', () => { const handleReads = (spy: ReturnType) => spy.mock.calls.filter(([sql]) => @@ -363,6 +427,26 @@ describe('the live cache across nodes', () => { ); }); + it('doesn`t cache a row a retire overtook during the app lookup', async () => { + const row = await store().mint( + input({ appUid: 'app-forget', keyPrefix: 'forget-race:' }), + ); + const apps = env.server.stores.app; + const getByUid = apps.getByUid.bind(apps); + const lookup = vi + .spyOn(apps, 'getByUid') + .mockImplementationOnce(async (...args) => { + await store().retireMany([row.handle], 'owner'); + return getByUid(...args); + }); + try { + await store().getLive(row.handle); + } finally { + lookup.mockRestore(); + } + expect(await store().getLive(row.handle)).toBeNull(); + }); + it('drops an owner\'s cached rows when the owner is suspended, here and on other nodes', async () => { const row = await store().mint( input({ appUid: 'app-forget', keyPrefix: 'forget-owner:' }), diff --git a/src/backend/stores/events/KvShareHandleStore.ts b/src/backend/stores/events/KvShareHandleStore.ts index 71a39699c..8635f6df6 100644 --- a/src/backend/stores/events/KvShareHandleStore.ts +++ b/src/backend/stores/events/KvShareHandleStore.ts @@ -24,12 +24,17 @@ import { mintKvHandleId, type KvShareAccess, } from '../../services/kvShare/kvShares.js'; +import { + APP_CLOCK_SKEW_SECONDS, + predatesApp, +} from '../../util/appGeneration.js'; import { BoundedTtlMap } from '../../util/boundedTtlMap.js'; import { keysetPage, openIdCursor, type PageResult, } from '../../util/pagination.js'; +import { KV_GLOBAL_APP_KEY } from '../systemKv/SystemKVStore.js'; import { PuterStore } from '../types.js'; import { assertColumnWidths, KV_SHARE_HANDLE_WIDTHS } from './columnWidths.js'; @@ -325,7 +330,8 @@ export class KvShareHandleStore extends PuterStore { * primary on a replica miss: that is for read-after-mint, and legacy ids * are no longer minted. A handle or owner forgotten in the last minute, * before or during the read, is read from the primary too. Concurrent - * misses on one handle share one read. + * misses on one handle share one read. A handle older than its app reads as + * gone. */ async getLive(handle: string): Promise { const kind = kvHandleIdKind(handle, this.config.jwt_secret_v2); @@ -365,8 +371,9 @@ export class KvShareHandleStore extends PuterStore { rows = await this.clients.db.pread(LIVE_QUERY, [handle]); } - // A forget that landed during the read may postdate what it returned. - const cacheable = + // A forget that landed during the read, or the app lookup after it, + // may postdate what it returned. + const cacheable = () => !forgottenSince(this.#forgotten.get(handle), before) && !forgottenSince( this.#forgottenOwners.get(Number(rows[0]?.owner_user_id)), @@ -375,19 +382,52 @@ export class KvShareHandleStore extends PuterStore { const found = rows[0]; if (!found || found.revoked_at !== null || !found.owner_uuid) { - if (cacheable) this.#missing.set(handle, true); + if (cacheable()) this.#missing.set(handle, true); return null; } const { label: _label, ...row } = toRow({ ...found, label: null }); + if (await this.predatesItsApp(row)) { + if (cacheable()) this.#missing.set(handle, true); + return null; + } const live: LiveKvShareHandle = { row, ownerUuid: String(found.owner_uuid), ownerSuspended: Boolean(Number(found.owner_suspended ?? 0)), }; - if (cacheable) this.#live.set(handle, live); + if (cacheable()) this.#live.set(handle, live); return live; } + /** + * Whether a handle was made for an earlier app under its app's uid: an + * origin-derived uid comes back verbatim when the app is registered again, + * and its predecessor's handles must not reach the new one. Such a handle + * reads as gone wherever a handle is resolved. + */ + async predatesItsApp( + row: Pick, + ): Promise { + if (row.appUid === KV_GLOBAL_APP_KEY) return false; + return predatesApp( + row.createdAt, + await this.stores.app.getByUid(row.appUid), + ); + } + + /** + * SQL that holds for a handle older than its app, the same rule as + * {@link predatesItsApp}. Binds one parameter, the clock-skew allowance. + */ + #predatesAppSql(): string { + const created = this.clients.db.unixSecondsSql('`apps`.`timestamp`'); + return ( + 'EXISTS (SELECT 1 FROM `apps` WHERE ' + + `\`apps\`.\`uid\` = \`${TABLE}\`.\`app_uid\` AND ` + + `${created} - ? > \`${TABLE}\`.\`created_at\`)` + ); + } + #forget(handle: string): void { this.#forgotten.set(handle, ++this.#forgets); this.#live.delete(handle); @@ -474,7 +514,8 @@ export class KvShareHandleStore extends PuterStore { * The live handle already minted over this exact region, if one exists — * what makes minting the same `(grantee, appUid, keyPrefix)` twice hand * back the one capability rather than a second row over the same - * permission. + * permission. One older than its app doesn't count, so sharing the region + * again makes a new one. */ async findLive( input: FindLiveKvShareHandleInput, @@ -490,12 +531,13 @@ export class KvShareHandleStore extends PuterStore { `SELECT ${SELECT_COLUMNS} FROM \`${TABLE}\` WHERE ` + '`owner_user_id` = ? AND `grantee_user_id` = ? AND ' + `\`app_uid\` = ? AND ${exactPrefix} AND \`revoked_at\` IS NULL ` + - 'LIMIT 1', + `AND NOT ${this.#predatesAppSql()} LIMIT 1`, [ input.ownerUserId, input.granteeUserId, input.appUid, input.keyPrefix, + APP_CLOCK_SKEW_SECONDS, ], ); return rows.length > 0 ? toRow(rows[0]) : null; @@ -622,19 +664,27 @@ export class KvShareHandleStore extends PuterStore { /** * One batch of live handles in these apps' namespaces, oldest first. A * deleted app's handles are retired off this; the caller repeats until a - * batch comes back short. + * batch comes back short. `createdAtMost` (unix seconds) leaves out handles + * made later, which belong to an app registered since under the same uid. */ async listLiveForApps( appUids: readonly string[], limit = KV_HANDLE_RETIRE_BATCH, + createdAtMost?: number, ): Promise { const unique = [...new Set(appUids)]; if (unique.length === 0) return []; + const created = + createdAtMost === undefined ? '' : 'AND `created_at` <= ? '; const rows = await this.clients.db.pread( `SELECT ${SELECT_COLUMNS} FROM \`${TABLE}\` ` + `WHERE \`app_uid\` IN (${unique.map(() => '?').join(', ')}) ` + - 'AND `revoked_at` IS NULL ORDER BY `id` LIMIT ?', - [...unique, limit], + `AND \`revoked_at\` IS NULL ${created}ORDER BY \`id\` LIMIT ?`, + [ + ...unique, + ...(createdAtMost === undefined ? [] : [createdAtMost]), + limit, + ], ); return rows.map(toRow); } @@ -642,22 +692,50 @@ export class KvShareHandleStore extends PuterStore { /** * One batch of live handles one owner holds out in one app's namespace, * oldest first, over the `(owner_user_id, app_uid, id)` index. The caller - * repeats until a batch comes back short. + * repeats until a batch comes back short. `createdAtMost` as in + * {@link listLiveForApps}. */ async listLiveForOwnerApp( ownerUserId: number, appUid: string, limit = KV_HANDLE_RETIRE_BATCH, + createdAtMost?: number, ): Promise { + const created = + createdAtMost === undefined ? '' : 'AND `created_at` <= ? '; const rows = await this.clients.db.pread( `SELECT ${SELECT_COLUMNS} FROM \`${TABLE}\` ` + 'WHERE `owner_user_id` = ? AND `app_uid` = ? ' + - 'AND `revoked_at` IS NULL ORDER BY `id` LIMIT ?', - [ownerUserId, appUid, limit], + `AND \`revoked_at\` IS NULL ${created}ORDER BY \`id\` LIMIT ?`, + [ + ownerUserId, + appUid, + ...(createdAtMost === undefined ? [] : [createdAtMost]), + limit, + ], ); return rows.map(toRow); } + /** + * One batch of live handles one user holds, by id after `afterId`, over the + * grantee index. Deleting the holder's account retires off this before its + * rows cascade away. + */ + async listLiveHeldBy( + granteeUserId: number, + afterId = 0, + limit = KV_HANDLE_RETIRE_BATCH, + ): Promise> { + const rows = await this.clients.db.pread( + `SELECT \`id\`, ${SELECT_COLUMNS} FROM \`${TABLE}\` ` + + 'WHERE `grantee_user_id` = ? AND `id` > ? ' + + 'AND `revoked_at` IS NULL ORDER BY `id` LIMIT ?', + [granteeUserId, afterId, limit], + ); + return rows.map((row) => ({ ...toRow(row), id: Number(row.id) })); + } + /** * Change a live handle's level or label in place, scoped to its owner. A * compare-and-set on `revoked_at IS NULL`, so a handle retired meanwhile @@ -697,7 +775,8 @@ export class KvShareHandleStore extends PuterStore { /** * Live handles held by one user, newest grants last, keyset-paginated on * the grantee index. Selected without the owner, prefix or grant string: - * this is the holder's listing, and none of those are theirs to see. + * this is the holder's listing, and none of those are theirs to see. One + * older than its app is left out. */ async listLiveForGrantee( granteeUserId: number, @@ -712,6 +791,8 @@ export class KvShareHandleStore extends PuterStore { filter.push('`app_uid` = ?'); filterParams.push(options.appUid); } + filter.push(`NOT ${this.#predatesAppSql()}`); + filterParams.push(APP_CLOCK_SKEW_SECONDS); const where = [...filter]; const params = [...filterParams]; if (after !== null) { @@ -771,6 +852,12 @@ export class KvShareHandleStore extends PuterStore { if (options.state === 'live') filter.push('`revoked_at` IS NULL'); if (options.state === 'revoked') filter.push('`revoked_at` IS NOT NULL'); + // A live one older than its app is left out until its retire lands + // and it lists as retired. + filter.push( + `(\`revoked_at\` IS NOT NULL OR NOT ${this.#predatesAppSql()})`, + ); + filterParams.push(APP_CLOCK_SKEW_SECONDS); const where = [...filter]; const params = [...filterParams]; diff --git a/src/backend/stores/systemKv/SystemKVStore.test.ts b/src/backend/stores/systemKv/SystemKVStore.test.ts index 82be9e866..b1d05e689 100644 --- a/src/backend/stores/systemKv/SystemKVStore.test.ts +++ b/src/backend/stores/systemKv/SystemKVStore.test.ts @@ -897,6 +897,107 @@ describe('SystemKVStore', () => { ).rejects.toThrow(); } }); + + it('asks stillWanted before deleting a page, and stops on a no', async () => { + await seed(3); + let asked = 0; + expect( + await target.purgeNamespace(namespaceOf(), { + maxItems: 10, + stillWanted: async () => { + asked++; + return false; + }, + }), + ).toEqual({ deleted: 0, done: false, stopped: true }); + expect(asked).toBe(1); + expect( + (await target.list({ as: 'keys' }, appOpts())).res, + ).toHaveLength(3); + + expect( + await target.purgeNamespace(namespaceOf(), { + maxItems: 10, + stillWanted: async () => true, + }), + ).toEqual({ deleted: 3, done: true }); + }); + }); + + describe('reserved items', () => { + const prefix = () => + `test-reserved#${Math.random().toString(36).slice(2)}#`; + + it('deletes one only while its condition holds', async () => { + const key = `${prefix()}item`; + const ifVersion = (version: number) => + target.deleteReservedItemIf( + key, + '#v = :v', + { ':v': version }, + { '#v': 'version' }, + ); + await target.putReservedItem(key, { version: 1 }); + + expect(await ifVersion(2)).toBe(false); + expect(await target.getReservedItem(key)).toMatchObject({ + version: 1, + }); + expect(await ifVersion(1)).toBe(true); + expect(await target.getReservedItem(key)).toBeNull(); + expect(await ifVersion(1)).toBe(false); + }); + + it('pages through them in key order from where the last page ended', async () => { + const shared = prefix(); + for (const name of ['a', 'b', 'c']) + await target.putReservedItem(`${shared}${name}`, { n: 1 }); + + const first = await target.pageReservedItems(shared, { limit: 2 }); + expect(first.items.map((item) => item.key)).toEqual([ + `${shared}a`, + `${shared}b`, + ]); + expect(first.nextKey).toBe(`${shared}b`); + const rest = await target.pageReservedItems(shared, { + limit: 2, + afterKey: first.nextKey, + }); + expect(rest.items.map((item) => item.key)).toEqual([`${shared}c`]); + expect(await target.queryReservedItems(shared)).toHaveLength(3); + }); + + it('updates one only while its condition holds, and returns it', async () => { + const key = `${prefix()}item`; + const claim = (owner: string) => + target.updateReservedItemIf( + key, + { set: { owner }, setIfAbsent: { since: 1 } }, + 'attribute_not_exists(#o)', + {}, + { '#o': 'owner' }, + ); + + expect(await claim('first')).toMatchObject({ + owner: 'first', + since: 1, + }); + expect(await claim('second')).toBeNull(); + expect( + await target.updateReservedItemIf( + key, + { set: { since: 2 }, remove: ['owner'] }, + '#o = :o', + { ':o': 'first' }, + { '#o': 'owner' }, + ), + ).toEqual( + expect.not.objectContaining({ owner: expect.anything() }), + ); + expect(await target.getReservedItem(key)).toMatchObject({ + since: 2, + }); + }); }); describe('expireAt / expire', () => { diff --git a/src/backend/stores/systemKv/SystemKVStore.ts b/src/backend/stores/systemKv/SystemKVStore.ts index 7a42033ac..70a27f77a 100644 --- a/src/backend/stores/systemKv/SystemKVStore.ts +++ b/src/backend/stores/systemKv/SystemKVStore.ts @@ -1659,19 +1659,38 @@ export class SystemKVStore extends PuterStore { async queryReservedItems>( prefix: string, ): Promise { + return (await this.pageReservedItems(prefix, { limit: 0 })).items; + } + + /** + * Up to `limit` reserved items whose key starts with `prefix`, in key order + * after `afterKey` (a `limit` of 0 reads one unbounded page). `nextKey` is + * where the next page starts, while more may remain. Expired rows are left + * out, as in {@link queryReservedItems}. + */ + async pageReservedItems>( + prefix: string, + { limit, afterKey }: { limit: number; afterKey?: string }, + ): Promise<{ items: T[]; nextKey?: string }> { const now = Date.now() / 1000; const response = await this.clients.dynamo.query( this.tableName, { namespace: SYSTEM_NAMESPACE }, - 0, - undefined, + limit, + afterKey === undefined + ? undefined + : { namespace: SYSTEM_NAMESPACE, key: afterKey }, '', false, { beginsWith: { key: 'key', value: prefix } }, ); - return ((response.Items ?? []) as T[]).filter( - (item) => !item.ttl || (item.ttl as number) > now, - ); + const last = response.LastEvaluatedKey?.key; + return { + items: ((response.Items ?? []) as T[]).filter( + (item) => !item.ttl || (item.ttl as number) > now, + ), + ...(last === undefined ? {} : { nextKey: String(last) }), + }; } /** @@ -1769,6 +1788,104 @@ export class SystemKVStore extends PuterStore { } } + /** + * Update one reserved item, creating it when missing, if `condition` holds. + * `set` attributes are written outright, `setIfAbsent` ones only where the + * item doesn't carry them, and `remove` ones are dropped. Returns the item + * as it now stands, or `null` when the condition refused it. Condition + * tokens must not start with `#u` or `:u`. + */ + async updateReservedItemIf>( + key: string, + change: { + set?: Record; + setIfAbsent?: Record; + remove?: string[]; + }, + condition: string, + conditionValues: Record = {}, + conditionNames: Record = {}, + ): Promise { + assertKey(key); + const names: Record = { ...conditionNames }; + const values: Record = { ...conditionValues }; + const setParts: string[] = []; + const removeParts: string[] = []; + let idx = 0; + const token = (attr: string) => { + const name = `#u${idx}`; + names[name] = attr; + return { name, value: `:u${idx++}` }; + }; + for (const [attr, value] of Object.entries(change.set ?? {})) { + const t = token(attr); + values[t.value] = value; + setParts.push(`${t.name} = ${t.value}`); + } + for (const [attr, value] of Object.entries(change.setIfAbsent ?? {})) { + const t = token(attr); + values[t.value] = value; + setParts.push(`${t.name} = if_not_exists(${t.name}, ${t.value})`); + } + for (const attr of change.remove ?? []) + removeParts.push(token(attr).name); + const expression = [ + setParts.length > 0 ? `SET ${setParts.join(', ')}` : '', + removeParts.length > 0 ? `REMOVE ${removeParts.join(', ')}` : '', + ] + .filter(Boolean) + .join(' '); + if (!expression) + throw new Error('kv: a reserved update needs a change'); + + try { + const response = await this.clients.dynamo.update( + this.tableName, + { namespace: SYSTEM_NAMESPACE, key }, + expression, + values, + names, + { condition }, + ); + return (response.Attributes as T | undefined) ?? null; + } catch (err) { + if (isConditionRefused(err)) return null; + throw err; + } + } + + /** + * Delete one reserved item, conditional on an attribute the caller read + * still holding. False means the item changed or is already gone. + */ + async deleteReservedItemIf( + key: string, + condition: string, + conditionValues: Record, + conditionNames: Record = {}, + ): Promise { + assertKey(key); + try { + await this.clients.dynamo.del( + this.tableName, + { namespace: SYSTEM_NAMESPACE, key }, + { + condition: { + expression: condition, + ...(Object.keys(conditionValues).length > 0 + ? { values: conditionValues } + : {}), + names: conditionNames, + }, + }, + ); + return true; + } catch (err) { + if (isConditionRefused(err)) return false; + throw err; + } + } + // -- Public API --------------------------------------------------- /** @@ -2765,12 +2882,17 @@ export class SystemKVStore extends PuterStore { /** * Delete up to `maxItems` items of one user's app namespace, as the * platform: no mutation events and no usage. Idempotent; `done` once the - * namespace is empty. + * namespace is empty. `stillWanted` is asked before each page is deleted, + * after its keys are read, and a no ends the call with `stopped`: what's + * written once it would say no is kept. */ async purgeNamespace( namespace: string, - { maxItems }: { maxItems: number }, - ): Promise<{ deleted: number; done: boolean }> { + { + maxItems, + stillWanted, + }: { maxItems: number; stillWanted?: () => Promise }, + ): Promise<{ deleted: number; done: boolean; stopped?: boolean }> { const parsed = parseKvNamespace(namespace); if ( !parsed || @@ -2787,7 +2909,9 @@ export class SystemKVStore extends PuterStore { const page = await this.#deletePage(namespace, pageKey, { limit: Math.min(maxItems - deleted, PURGE_PAGE_ITEMS), consistent: true, + beforeDelete: stillWanted, }); + if (page.stopped) return { deleted, done: false, stopped: true }; deleted += page.deleted; pageKey = page.pageKey; if (!pageKey) return { deleted, done: true }; @@ -2804,11 +2928,20 @@ export class SystemKVStore extends PuterStore { async #deletePage( namespace: string, pageKey: Record | undefined, - { limit = 0, consistent }: { limit?: number; consistent: boolean }, + { + limit = 0, + consistent, + beforeDelete, + }: { + limit?: number; + consistent: boolean; + beforeDelete?: () => Promise; + }, ): Promise<{ deleted: number; usage: KVUsage; pageKey?: Record; + stopped?: boolean; }> { const response = await this.clients.dynamo.query( this.tableName, @@ -2823,6 +2956,8 @@ export class SystemKVStore extends PuterStore { response.ConsumedCapacity?.CapacityUnits as number | undefined, ); const keys = (response.Items ?? []).map((entry) => String(entry.key)); + if (keys.length > 0 && beforeDelete && !(await beforeDelete())) + return { deleted: 0, usage, stopped: true }; if (keys.length > 0) { try { const deleted = await this.clients.dynamo.batchDel( diff --git a/src/backend/util/appGeneration.test.ts b/src/backend/util/appGeneration.test.ts new file mode 100644 index 000000000..d3d78dcb0 --- /dev/null +++ b/src/backend/util/appGeneration.test.ts @@ -0,0 +1,46 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import { describe, expect, it } from 'vitest'; +import { + APP_CLOCK_SKEW_SECONDS, + appGenerationStart, + predatesApp, +} from './appGeneration.js'; + +describe('app generations', () => { + const app = { created_epoch: 10_000 }; + + it('starts a clock-skew allowance before the app row was created', () => { + expect(appGenerationStart(app)).toBe(10_000 - APP_CLOCK_SKEW_SECONDS); + }); + + it('has no start without a creation time', () => { + for (const row of [null, undefined, {}, { created_epoch: 0 }]) + expect(appGenerationStart(row)).toBeNull(); + }); + + it('counts only records made before that start as older', () => { + const start = 10_000 - APP_CLOCK_SKEW_SECONDS; + expect(predatesApp(start - 1, app)).toBe(true); + expect(predatesApp(start, app)).toBe(false); + expect(predatesApp(10_000 + 5, app)).toBe(false); + expect(predatesApp(1, null)).toBe(false); + }); +}); diff --git a/src/backend/util/appGeneration.ts b/src/backend/util/appGeneration.ts new file mode 100644 index 000000000..2f22d02fc --- /dev/null +++ b/src/backend/util/appGeneration.ts @@ -0,0 +1,53 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +// An origin-derived app uid comes back verbatim when its app is registered +// again, so anything keyed on the uid alone can outlive the app it was for. + +/** + * Allowed gap between a record's creation time (server clock) and its app row's + * (possibly the database clock) before the record counts as older. + */ +export const APP_CLOCK_SKEW_SECONDS = 60; + +/** An app row as read from the store, with its creation time. */ +export interface AppGenerationRow { + created_epoch?: unknown; +} + +/** + * Unix seconds before which a record tied to `app`'s uid belongs to an earlier + * app with the same uid, or `null` when the creation time is unknown. + */ +export const appGenerationStart = ( + app: AppGenerationRow | null | undefined, +): number | null => { + const created = Number(app?.created_epoch); + if (!Number.isFinite(created) || created <= 0) return null; + return created - APP_CLOCK_SKEW_SECONDS; +}; + +/** Whether a record made at `createdAt` (unix seconds) predates `app`. */ +export const predatesApp = ( + createdAt: number, + app: AppGenerationRow | null | undefined, +): boolean => { + const start = appGenerationStart(app); + return start !== null && createdAt > 0 && createdAt < start; +}; diff --git a/src/docs/src/Events/subjects.md b/src/docs/src/Events/subjects.md index 8ec3dfff7..f153d87bf 100644 --- a/src/docs/src/Events/subjects.md +++ b/src/docs/src/Events/subjects.md @@ -126,7 +126,7 @@ await puter.events.onLocal(`kv:${handle}:*`, ({ event }) => render(event.key)); - **Keys can't leave the handle.** A handle with no key after it, or a key using `..`, is refused with `invalid_kv_handle_key`. - **Writes from every side arrive.** A change the owner makes, and a change any holder makes through a `write` share, reach the owner's subscriptions and every holder's. On a holder's own write through the handle, `self` is `true`, so `self && !viaWorker` skips your own changes. See [the event](#the-event). -When a share stops, deliveries through its handle stop, and every persistent subscription on it is suspended with `permission_revoked` and its backlog dropped. Every [`onLocal()`](/Events/onLocal/) subscription on it ends, and its `onError` gets `subscription_ended` with `permission_revoked`. A connection to another region can take up to 20 minutes to be told, and gets no deliveries through the handle in the meantime. If another share to the same holder still covers its keys, live deliveries through the stopped handle can continue for up to 60 seconds. This happens when the owner revokes it with [`puter.kv.unshare()`](/KV/unshare/), when the holder removes their own access the same way, and when the app whose data it is gets deleted. Where `watch` shares are off on the deployment, handle subjects fail with `events_kv_handles_disabled`. +When a share stops, deliveries through its handle stop, and every persistent subscription on it is suspended with `permission_revoked` and its backlog dropped. Every [`onLocal()`](/Events/onLocal/) subscription on it ends, and its `onError` gets `subscription_ended` with `permission_revoked`. A connection to another region can take up to 20 minutes to be told, and gets no deliveries through the handle in the meantime. If another share to the same holder still covers its keys, live deliveries through the stopped handle can continue for up to 60 seconds. This happens when the owner revokes it with [`puter.kv.unshare()`](/KV/unshare/), when the holder removes their own access the same way, when the app whose data it is gets deleted, when the owner removes that app along with its data, and when the holder deletes their account. Where `watch` shares are off on the deployment, handle subjects fail with `events_kv_handles_disabled`. **Apps.** An app can share, list, and unshare its user's data with [`puter.kv.share()`](/KV/share/) and its sibling methods, but only inside its own namespace and with the user's consent. To ask for it, call [`puter.perms.request()`](/Perms/request/) with the `'kvShare'` resource, the prefix the shares will cover, and the highest level they'll have: diff --git a/src/docs/src/KV/listSharedByMe.md b/src/docs/src/KV/listSharedByMe.md index 9a4778c1c..afba22590 100644 --- a/src/docs/src/KV/listSharedByMe.md +++ b/src/docs/src/KV/listSharedByMe.md @@ -40,7 +40,7 @@ A `Promise` that resolves to: - `label` (String | null): The label the holder is shown. - `createdAt` (Number): When the share was made, in Unix seconds. - `revokedAt` (Number | null): When the share stopped, in Unix seconds, or `null` while it's live. - - `revokedBy` (String | null): Who stopped it. It's `'owner'` when you revoked it, `'holder'` when the holder removed their own access, and `'system'` when the app whose data it was got deleted. It's `null` while the share is live, and on some shares that stopped before Puter recorded who stopped them. + - `revokedBy` (String | null): Who stopped it. It's `'owner'` when you revoked it, `'holder'` when the holder removed their own access, and `'system'` when the app whose data it was got deleted or you removed that app along with its data. It's `null` while the share is live, and on some shares that stopped before Puter recorded who stopped them. - `cursor` (String): Pass it as `cursor` to get the next page. It's included only if there are more pages to load. - `total` (Number): The number of shares across every page that match `appUid` and `state`. Only included when `includeTotal` is `true`. diff --git a/src/docs/src/KV/share.md b/src/docs/src/KV/share.md index 6f3818f86..3748986e7 100644 --- a/src/docs/src/KV/share.md +++ b/src/docs/src/KV/share.md @@ -87,6 +87,8 @@ Sharing the same prefix of the same app's data with the same user again changes Leave out `label` to keep the current label, pass a new one to replace it, or pass `null` to clear it. +Concurrent calls that share the same prefix with the same user usually make one share, and every call resolves to it with the `access` and `label` of the call that made it. This isn't guaranteed for calls that reach different Puter regions at the same moment, which can each make a share. [`puter.kv.listSharedByMe()`](/KV/listSharedByMe/) lists both, and you can stop the extra one with [`puter.kv.unshare()`](/KV/unshare/). + ## What the holder sees The holder is never told who shared or what the prefix is. The keys they read and write are relative to the prefix. The label is the only thing about the share you choose to tell them. diff --git a/src/puter-js/tests/api/suites/perms.suite.ts b/src/puter-js/tests/api/suites/perms.suite.ts index bb8110c9f..4ca19bf5f 100644 --- a/src/puter-js/tests/api/suites/perms.suite.ts +++ b/src/puter-js/tests/api/suites/perms.suite.ts @@ -82,12 +82,13 @@ export default suite('perms', { }); t.assert.ok(!revoked.error, `revoke failed: ${JSON.stringify(revoked)}`); - // The data is deleted in the background after the call returns. + // The data is deleted in the background after the call returns. Read + // as the account: opening the app again would stop the deletion. let value: unknown = 'x'; for (let i = 0; i < 50 && value !== null; i++) { - value = await asApp(t, app.uid, () => - t.puter.kv.get('perms-suite-key'), - ); + value = await t.puter.kv.get('perms-suite-key', { + appUuid: app.uid, + }); if (value !== null) await new Promise((r) => setTimeout(r, 100)); } t.assert.equal(value, null);