diff --git a/docs/contributing/architecture/entitlements.md b/docs/contributing/architecture/entitlements.md index 110df6eb6c..a3807b2c4c 100644 --- a/docs/contributing/architecture/entitlements.md +++ b/docs/contributing/architecture/entitlements.md @@ -21,7 +21,9 @@ Module: `packages/worker/src/entitlements/` helpers for rate-style limits, `assertWithinStorageBytesEntitlement` (UserMeter DO reserve with cold bootstrap), and `readCurrentEntitlementResourceUsage` (UserMeter-authoritative for - `storage_bytes` and daily resources). + `storage_bytes`, `package_services`, and daily resources). + `countRunningPackageServices` is UserMeter-authoritative. + `countRunningPackageServicesFromD1` is reserved for parity only. ## Plan model @@ -130,11 +132,11 @@ live 2026-08-01). `assertWithinStorageBytesEntitlement` uses atomic DO async mirror** — it is never read for enforcement or usage after the flip; only parity checks, reconcile tooling, and cold bootstrap read it. -**D1 package service liveness** (`package_service_states`) stays authoritative -for running counts, discovery, and `service_start` enforcement. UserMeter schema -v5 adds an optional per-service shadow table as **best-effort future-cutover -support only** — see -[Package service liveness — UserMeter shadow](#package-service-liveness--usermeter-shadow-expand-phase-slice-4-phase-a). +**D1 package service liveness** (`package_service_states`): UserMeter is now the +**authoritative running-count source** for `package_services` enforcement and +`service_start` after the cutover (2026-08-01). D1 remains the enumeration index +for discovery, account export, and deletion, and is the parity mirror — see +[Package service liveness — UserMeter authority](#package-service-liveness--usermeter-authority-cutover-2026-08-01). **Account-deletion write fencing** uses a split authority model after slice 5 Phase B: D1 `users.deleting_at` remains the permanent point gate; callers that @@ -246,22 +248,71 @@ return `null` for each shadow so paged consumers never double-count them `UserMeter.purge()` clears counters, inbound delivery claims, and all shadow state (storage bytes and package-service liveness) via `deleteAll`. -### Package service liveness — UserMeter shadow (expand phase slice 4, Phase A) - -D1 `package_service_states` remains the **sole authority** for running-service -**count**, **discovery**, and **`service_start` enforcement** in this PR. -Nothing in Phase A switches those reads or the `assertWithinEntitlement` path -for `package_services` / `persistent_package_services`. - -UserMeter schema **v5** adds a per-service `package_service_states` table inside -the DO as **best-effort shadow / future-cutover support only** (`status`, -`started_at`, monotonic `source_updated_at` from the D1 projection timestamp, -`revision`, `updated_at`). User scope is the DO identity — there is no `user_id` -column. Shadow rows are never read for usage display, entitlement enforcement, -or account-deletion inventory in this slice. - -**Dual-write from `PackageServiceInstance`:** every D1 projection also attempts -a best-effort UserMeter shadow on the same lifecycle surface: +### Package service liveness — UserMeter authority (cutover 2026-08-01) + +**Production evidence (authority gate):** 38-user parity sweep at 2026-08-01 +~21:27Z showed running-count parity clean for every user. Exactly one user had 2 +D1-only stopped inventory rows with `d1-fresh-running=0` and `meter-running=0`. +No declared package services exist in production (Kent's 59 personal or 10 +platform packages), so deliberate lifecycle exercise is unavailable; the parity +sweep is the production gate. + +**Authority after the flip:** `countRunningPackageServices` reads directly from +the UserMeter DO — no D1 access on the enforcement path. New lifecycle +dual-writes (`upsertPackageServiceState`) populate the meter so enforcement is +immediately consistent. D1 `package_service_states` remains: + +- the **enumeration index** for account export, deletion, and admin discovery +- the **parity mirror**: `admin_user_meter_parity` compares + `countRunningPackageServicesFromD1` (D1 side) against + `meter.countRunningPackageServices` (DO side) +- the **repair source**: `bootstrapPackageServiceStates` is an explicit + migration/repair primitive only — it is not on the enforcement path + +**Stopped inventory policy:** D1 stopped/error rows are historical inventory and +do not gate liveness authority. They are never loaded into the meter at +enforcement time. The meter count is 0 for a new user or a user with no live +dual-writes; this is correct. + +**D1-only fresh-running row:** a D1 fresh-running row that has no corresponding +meter row means a lifecycle dual-write shadow missed the meter. This is a +migration anomaly and parity blocker; `admin_user_meter_parity` will expose it. +The authority count returns 0 (from the empty meter). Roll back D1 authority if +this is observed in production. + +**`excludeService` and 24h staleness:** both semantics are preserved exactly. +`excludeService` is forwarded to `meter.countRunningPackageServices`. The 24h +`staleAfterMs` window applies to DO `source_updated_at` (kept fresh by heartbeat +dual-writes). + +**`readEntitlementResourceUsage` throws:** calling +`readEntitlementResourceUsage(package_services)` throws with guidance directing +callers to `readCurrentEntitlementResourceUsage` or `assertWithinEntitlement` +with an explicit `getCurrent`. No production D1 count path for enforcement +remains. + +**`service_start` caller:** passes `env: ctx.env` into +`countRunningPackageServices` so `USER_METER` binding is available. Missing +binding fails closed. + +**Rollback window:** D1 `package_service_states` continues to receive D1-first +lifecycle dual-writes (no table/column retirements). To roll back: remove the +`readCurrentEntitlementResourceUsage` / `countRunningPackageServices` changes, +revert `service-start.ts` to call `countRunningPackageServicesFromD1`, and +restore the `readEntitlementResourceUsage` D1 path. Verify via +`admin_user_meter_parity` before rolling back. + +**No declared service lifecycle:** because no package services are currently +declared in production packages, the only ongoing exercise of this path is the +heartbeat alarm and any future `service_start` calls. Parity sweeps remain the +primary verification surface. + +UserMeter schema **v5** adds the per-service `package_service_states` table +inside the DO (`status`, `started_at`, monotonic `source_updated_at`, +`revision`, `updated_at`). + +**Dual-write from `PackageServiceInstance`:** D1 upsert/delete runs first; a +best-effort UserMeter shadow follows on the same lifecycle surfaces: - lifecycle transitions and warm-start restore after upgrades (`projectServiceStateToD1`) @@ -271,27 +322,17 @@ a best-effort UserMeter shadow on the same lifecycle surface: - purge (`deleteProjectedServiceState` deletes D1 then shadow before `deleteAll`) -D1 upsert/delete runs first; shadow RPCs are optional when `USER_METER` is -unbound and failures log `package-service-user-meter-shadow-failed` without -affecting the service path. Shadow upserts reject stale/out-of-order writes when -`sourceUpdatedAt` is older than the existing shadow row so cold bootstrap cannot -clobber fresher state. - -**Timing unchanged:** live services heartbeat D1 `updated_at` every **1 hour** -(`packageServiceStateHeartbeatMs`). Running counts still treat rows as stale -after **24 hours** without a fresh heartbeat (`packageServiceStateStaleMs` in -`entitlements/service.ts`). The UserMeter cutover-support RPC -`countRunningPackageServices` uses the same 24h window on shadow -`source_updated_at` but is **not** wired to enforcement in Phase A. +Shadow RPCs are optional when `USER_METER` is unbound and failures log +`package-service-user-meter-shadow-failed` without affecting the service path. +Shadow upserts reject stale/out-of-order writes when `sourceUpdatedAt` is older +than the existing shadow row. **Account export:** `UserMeter.exportCounters` returns additive `packageServiceStatesShadow` on the first page only (`startAfter` absent); later -pages return `null`. Section totals count the shadow inventory once when -present; the field is explicitly non-authoritative — authoritative liveness -remains on D1. +pages return `null`. The field reflects the authoritative DO inventory. -**Account purge:** `UserMeter.purge()` clears package-service shadow rows with -the rest of DO state via `deleteAll`. +**Account purge:** `UserMeter.purge()` clears package-service shadow rows via +`deleteAll` then schema re-init. ### Account-deletion write fencing — UserMeter authority (expand phase slice 5, Phase B) @@ -351,19 +392,14 @@ restores any deleting tombstone; D1 `deleting_at` remains the gate. Public errors, ALS nested-lease reuse, holder strings, and `waitUntil` release detachment stay parity-compatible with the pre-cutover surface. -### Future package-service authority flip (contract follow-up) - -A separate **high-risk contract PR** — not a merge blocker for Phase A — will -flip running-service count/discovery/enforcement into UserMeter only after: - -1. at least one full **24h stale-window soak** with shadow/D1 parity review, and -2. a **cold-bootstrap design** for accounts whose DO shadow is empty while D1 - still holds rows (`bootstrapPackageServiceStates` / equivalent). +### Package-service authority flip (complete, 2026-08-01) -Until that flip, shadow divergence is acceptable; D1 remains the contract. **D1 -likely stays the enumeration index** for account export, deletion, and admin -discovery until an alternate inventory exists — UserMeter would become the -running-count authority first, not a wholesale replacement for every D1 reader. +The running-count authority for `package_services` moved from D1 to UserMeter as +of the cutover (see +[Package service liveness — UserMeter authority](#package-service-liveness--usermeter-authority-cutover-2026-08-01)). +D1 remains the enumeration index for account export, deletion, and admin +discovery. `countRunningPackageServicesFromD1` is retained for parity +comparisons; all enforcement uses UserMeter via `countRunningPackageServices`. ### Storage authority flip (complete, 2026-08-01) @@ -622,14 +658,16 @@ Rules: ## Counting strategy - **Row-count limits** (saved packages, scheduled jobs, repo sessions, secrets, - running package services) are counted **directly from their source D1 tables - at the enforcement point** via built-in counters in `service.ts`. They do not - depend on any metering or rollup tables. Running package services are counted - from D1 `package_service_states` (status `running` and freshly heartbeaten; 1h - heartbeat, 24h staleness), not from run-history rows — see - [Run records](./run-records.md) (`state-vs-history`). Expand-phase slice 4 - Phase A dual-writes the same projection into UserMeter as a non-authoritative - shadow; enforcement and `service_start` still read D1 only. **Concurrent + running package services) are counted via built-in counters in `service.ts`. + Most are counted directly from their source D1 tables. **Running package + services** are now counted from the **per-user UserMeter DO** (authoritative + as of 2026-08-01): `countRunningPackageServices` counts `status = 'running'` + rows with the 24h staleness window from DO `source_updated_at`. D1 + `package_service_states` remains the enumeration index (discovery, export, + deletion) and parity mirror; `countRunningPackageServicesFromD1` is retained + for parity only — see + [Package service liveness — UserMeter authority](#package-service-liveness--usermeter-authority-cutover-2026-08-01) + and [Run records](./run-records.md) (`state-vs-history`). **Concurrent workflows** are authoritative in per-user RunLog `workflow_projections`: create reserves atomically via `reserveWorkflowProjectionSlot`, and usage readers call `countActiveWorkflowProjections` through @@ -782,7 +820,7 @@ Use `getCurrent` only when the built-in D1 counter cannot express the resource. | ----------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `scheduled_jobs` | `createJob` in `packages/worker/src/jobs/service.ts` (exemplar) | | `saved_packages` | new-package branch of `package_save` and projection insert | -| `package_services` | `service_start` capability path (count from `package_service_states`) | +| `package_services` | `service_start` capability path (`getCurrent` with `countRunningPackageServices(env)`; UserMeter authority since 2026-08-01; D1-only fresh-running rows are parity blockers rather than runtime bootstrap input) | | `persistent_package_services` | `service_start` for services declared `mode: 'persistent'` | | `repo_sessions` | `repo_open_session` before creating a new session | | `email_sends_per_day` | `sendOutboundEmail` (`consumeDailyEntitlement`; plan limit from `resolvePlanLimit`) | diff --git a/packages/worker/src/admin/user-meter-parity.ts b/packages/worker/src/admin/user-meter-parity.ts index 6cf77d1c6a..8e6691cabd 100644 --- a/packages/worker/src/admin/user-meter-parity.ts +++ b/packages/worker/src/admin/user-meter-parity.ts @@ -10,7 +10,7 @@ import { type UserMeterWriteLeaseShadow, } from '#worker/entitlements/user-meter-do.ts' import { - countRunningPackageServices, + countRunningPackageServicesFromD1, packageServiceStateStaleMs, } from '#worker/entitlements/service.ts' import { planLimits } from '#worker/entitlements/plans.ts' @@ -462,7 +462,7 @@ async function readPackageServicesParity(input: { d1Rows, meterStates, }) - const d1FreshRunningCount = await countRunningPackageServices({ + const d1FreshRunningCount = await countRunningPackageServicesFromD1({ db: input.db, userId: input.stableUserId, now: input.now, diff --git a/packages/worker/src/entitlements/entitlements.node.test.ts b/packages/worker/src/entitlements/entitlements.node.test.ts index b007f28ee7..5c4ba17fc6 100644 --- a/packages/worker/src/entitlements/entitlements.node.test.ts +++ b/packages/worker/src/entitlements/entitlements.node.test.ts @@ -23,7 +23,7 @@ import { assertWithinEntitlement, assertWithinStorageBytesEntitlement, consumeDailyEntitlement, - countRunningPackageServices, + countRunningPackageServicesFromD1, estimateEntitlementStorageEntryByteDelta, estimateEntitlementStorageEntryBytes, findCachedUserAccountByStableUserId, @@ -1613,7 +1613,7 @@ test('getUserPlan resolves effective plan from manual plan and stripe_plan', asy ).toBe('max') }) -test('countRunningPackageServices queries package_service_states with staleness and excludeService', async () => { +test('countRunningPackageServicesFromD1 queries package_service_states with staleness and excludeService', async () => { const now = new Date('2026-07-26T12:00:00.000Z') const queries: Array<{ sql: string; params: Array }> = [] const db = { @@ -1632,7 +1632,7 @@ test('countRunningPackageServices queries package_service_states with staleness } as unknown as D1Database expect( - await countRunningPackageServices({ + await countRunningPackageServicesFromD1({ db, userId: 'user-1', now, @@ -1645,7 +1645,7 @@ test('countRunningPackageServices queries package_service_states with staleness ]) queries.length = 0 - await countRunningPackageServices({ + await countRunningPackageServicesFromD1({ db, userId: 'user-1', now, diff --git a/packages/worker/src/entitlements/package-service-states.workers.test.ts b/packages/worker/src/entitlements/package-service-states.workers.test.ts index b011d76cba..3c00803cbe 100644 --- a/packages/worker/src/entitlements/package-service-states.workers.test.ts +++ b/packages/worker/src/entitlements/package-service-states.workers.test.ts @@ -2,7 +2,7 @@ import { env } from 'cloudflare:workers' import { expect, test } from 'vitest' import { assertWithinEntitlement, - countRunningPackageServices, + countRunningPackageServicesFromD1, packageServiceStateStaleMs, } from '#worker/entitlements/service.ts' import { EntitlementLimitError } from '#worker/entitlements/errors.ts' @@ -100,14 +100,14 @@ test('package_service_states counts fresh running rows and enforces the plan bou }) expect( - await countRunningPackageServices({ + await countRunningPackageServicesFromD1({ db: env.APP_DB, userId, now, }), ).toBe(1) expect( - await countRunningPackageServices({ + await countRunningPackageServicesFromD1({ db: env.APP_DB, userId, now, @@ -124,7 +124,7 @@ test('package_service_states counts fresh running rows and enforces the plan bou updatedAt: fresh, }) expect( - await countRunningPackageServices({ + await countRunningPackageServicesFromD1({ db: env.APP_DB, userId, now, @@ -153,12 +153,20 @@ test('package_service_states counts fresh running rows and enforces the plan bou }) } + // assertWithinEntitlement for package_services requires getCurrent with the + // authoritative counter (enforcement uses UserMeter; parity/bootstrap uses D1). const denied = await assertWithinEntitlement({ db: env.APP_DB, userId: limitUserId, email, resource: 'package_services', now, + getCurrent: async () => + await countRunningPackageServicesFromD1({ + db: env.APP_DB, + userId: limitUserId, + now, + }), }).then( () => null, (thrown: unknown) => thrown, @@ -182,7 +190,7 @@ test('package_service_states counts fresh running rows and enforces the plan bou resource: 'package_services', now, getCurrent: async () => - await countRunningPackageServices({ + await countRunningPackageServicesFromD1({ db: env.APP_DB, userId: limitUserId, now, diff --git a/packages/worker/src/entitlements/package-services-user-meter.node.test.ts b/packages/worker/src/entitlements/package-services-user-meter.node.test.ts new file mode 100644 index 0000000000..a055857629 --- /dev/null +++ b/packages/worker/src/entitlements/package-services-user-meter.node.test.ts @@ -0,0 +1,223 @@ +import { expect, test } from 'vitest' +import { createStableUserIdFromEmail } from '#worker/user-id.ts' +import { createInMemoryUserMeterEnv } from '#worker/test-support/user-meter.ts' +import { type UserMeterRpc } from '#worker/entitlements/user-meter-client.ts' +import { + countRunningPackageServices, + packageServiceStateStaleMs, +} from './service.ts' + +type SeedInput = { + packageId: string + serviceName: string + status: 'running' | 'stopped' | 'error' + sourceUpdatedAt: string + startedAt?: string | null +} + +async function seedService( + env: { USER_METER?: DurableObjectNamespace }, + userId: string, + input: SeedInput, +) { + const meter = env.USER_METER!.get( + env.USER_METER!.idFromName(userId), + ) as unknown as UserMeterRpc + await meter.upsertPackageServiceState({ + packageId: input.packageId, + serviceName: input.serviceName, + status: input.status, + startedAt: input.startedAt ?? null, + sourceUpdatedAt: input.sourceUpdatedAt, + }) +} + +test('D1-only stopped rows are ignored: empty meter returns zero', async () => { + // Policy: D1 stopped/error rows are historical inventory and do not gate + // liveness authority. countRunningPackageServices never reads D1; it reads + // only the meter. An empty meter returns 0 even if D1 has stopped rows. + const userId = await createStableUserIdFromEmail( + 'stopped-ignored@example.com', + ) + const now = new Date('2026-08-01T21:27:00.000Z') + const { env } = createInMemoryUserMeterEnv() + + const count = await countRunningPackageServices({ env, userId, now }) + expect(count).toBe(0) +}) + +test('D1-only fresh-running row is ignored by authority count — parity blocker', async () => { + // Policy: a D1-only fresh-running row means a lifecycle dual-write shadow + // missed the meter. This is a migration anomaly / parity blocker. The + // authority count sees 0 (the meter is empty). Observe via + // admin_user_meter_parity; roll back D1 authority if seen in production. + const userId = await createStableUserIdFromEmail( + 'd1-only-running@example.com', + ) + const now = new Date('2026-08-01T21:27:00.000Z') + const { env } = createInMemoryUserMeterEnv() + + // countRunningPackageServices never consults D1; empty meter → 0. + const count = await countRunningPackageServices({ env, userId, now }) + expect(count).toBe(0) +}) + +test('live meter fresh running rows are counted', async () => { + const userId = await createStableUserIdFromEmail('fresh-running@example.com') + const now = new Date('2026-08-01T21:27:00.000Z') + const freshAt = now.toISOString() + const { env } = createInMemoryUserMeterEnv() + + await seedService(env, userId, { + packageId: 'pkg-a', + serviceName: 'alpha', + status: 'running', + startedAt: freshAt, + sourceUpdatedAt: freshAt, + }) + await seedService(env, userId, { + packageId: 'pkg-b', + serviceName: 'beta', + status: 'running', + startedAt: freshAt, + sourceUpdatedAt: freshAt, + }) + + const count = await countRunningPackageServices({ env, userId, now }) + expect(count).toBe(2) +}) + +test('stale running row does not count against the limit', async () => { + const userId = await createStableUserIdFromEmail('stale-rows@example.com') + const now = new Date('2026-08-01T21:27:00.000Z') + const staleAt = new Date( + now.valueOf() - packageServiceStateStaleMs - 1, + ).toISOString() + const freshAt = now.toISOString() + const { env } = createInMemoryUserMeterEnv() + + await seedService(env, userId, { + packageId: 'pkg-stale', + serviceName: 'stale-svc', + status: 'running', + startedAt: staleAt, + sourceUpdatedAt: staleAt, + }) + await seedService(env, userId, { + packageId: 'pkg-fresh', + serviceName: 'fresh-svc', + status: 'running', + startedAt: freshAt, + sourceUpdatedAt: freshAt, + }) + + const count = await countRunningPackageServices({ env, userId, now }) + expect(count).toBe(1) +}) + +test('excludeService removes the specified service from the running count', async () => { + const userId = await createStableUserIdFromEmail('exclude-svc@example.com') + const now = new Date('2026-08-01T21:27:00.000Z') + const freshAt = now.toISOString() + const { env } = createInMemoryUserMeterEnv() + + await seedService(env, userId, { + packageId: 'pkg-a', + serviceName: 'alpha', + status: 'running', + startedAt: freshAt, + sourceUpdatedAt: freshAt, + }) + await seedService(env, userId, { + packageId: 'pkg-b', + serviceName: 'beta', + status: 'running', + startedAt: freshAt, + sourceUpdatedAt: freshAt, + }) + + const withExclusion = await countRunningPackageServices({ + env, + userId, + now, + excludeService: { packageId: 'pkg-a', serviceName: 'alpha' }, + }) + expect(withExclusion).toBe(1) + + const withoutExclusion = await countRunningPackageServices({ + env, + userId, + now, + }) + expect(withoutExclusion).toBe(2) +}) + +test('empty meter (new user) returns zero', async () => { + const userId = await createStableUserIdFromEmail('empty-user@example.com') + const now = new Date('2026-08-01T21:27:00.000Z') + const { env } = createInMemoryUserMeterEnv() + + const count = await countRunningPackageServices({ env, userId, now }) + expect(count).toBe(0) +}) + +test('USER_METER binding missing throws fail-closed', async () => { + const userId = await createStableUserIdFromEmail('no-binding@example.com') + const now = new Date('2026-08-01T21:27:00.000Z') + const emptyEnv = {} as { USER_METER?: DurableObjectNamespace } + + await expect( + countRunningPackageServices({ env: emptyEnv, userId, now }), + ).rejects.toThrow('USER_METER Durable Object binding is not configured.') +}) + +test('meter count is stable across calls (D1 drift is irrelevant)', async () => { + // Authority is the meter only. Repeated calls return consistent results + // regardless of any D1 state — D1 is never consulted. + const userId = await createStableUserIdFromEmail('stable-meter@example.com') + const now = new Date('2026-08-01T21:27:00.000Z') + const freshAt = now.toISOString() + const { env } = createInMemoryUserMeterEnv() + + await seedService(env, userId, { + packageId: 'pkg-a', + serviceName: 'alpha', + status: 'running', + startedAt: freshAt, + sourceUpdatedAt: freshAt, + }) + + const first = await countRunningPackageServices({ env, userId, now }) + const second = await countRunningPackageServices({ env, userId, now }) + expect(first).toBe(1) + expect(second).toBe(1) +}) + +test('lifecycle dual-write (upsertPackageServiceState) populates meter for enforcement', async () => { + // Verify that shadow writes from lifecycle transitions are visible to the + // authority count without any D1 bootstrap. + const userId = await createStableUserIdFromEmail('shadow-write@example.com') + const now = new Date('2026-08-01T21:27:00.000Z') + const freshAt = now.toISOString() + const { env } = createInMemoryUserMeterEnv() + + expect(await countRunningPackageServices({ env, userId, now })).toBe(0) + + await seedService(env, userId, { + packageId: 'pkg-live', + serviceName: 'live-svc', + status: 'running', + startedAt: freshAt, + sourceUpdatedAt: freshAt, + }) + expect(await countRunningPackageServices({ env, userId, now })).toBe(1) + + await seedService(env, userId, { + packageId: 'pkg-live', + serviceName: 'live-svc', + status: 'stopped', + startedAt: null, + sourceUpdatedAt: new Date(now.valueOf() + 1000).toISOString(), + }) + expect(await countRunningPackageServices({ env, userId, now })).toBe(0) +}) diff --git a/packages/worker/src/entitlements/service.ts b/packages/worker/src/entitlements/service.ts index 79ed0f4233..07df5e1cdb 100644 --- a/packages/worker/src/entitlements/service.ts +++ b/packages/worker/src/entitlements/service.ts @@ -794,13 +794,11 @@ export async function markUserD1StorageReconciliationAttempt(input: { export const packageServiceStateStaleMs = 24 * 60 * 60 * 1000 /** - * Count currently-running package services for a user from - * `package_service_states`. Enforcement points that start a specific service - * should pass `excludeService` so a stale 'running' row for that same - * service can never block its own restart (starting it again does not add a - * new running service). + * Count currently-running package services for a user directly from D1 + * `package_service_states`. Reserved for parity comparisons only — enforcement + * uses the UserMeter-authoritative `countRunningPackageServices`. */ -export async function countRunningPackageServices(input: { +export async function countRunningPackageServicesFromD1(input: { db: D1Database userId: string excludeService?: { packageId: string; serviceName: string } @@ -832,6 +830,32 @@ export async function countRunningPackageServices(input: { ) } +/** + * UserMeter-authoritative running count for a user's package services. + * Reads directly from the DO; no D1 access. Lifecycle dual-writes + * (`upsertPackageServiceState`) populate the meter so enforcement is + * immediately consistent. The `excludeService` and 24h staleness semantics + * are preserved exactly. + * + * Requires `env.USER_METER`. Throws immediately when the binding is absent + * (fail-closed for enforcement). + */ +export async function countRunningPackageServices(input: { + env: UserMeterEnv + userId: string + excludeService?: { packageId: string; serviceName: string } + now?: Date +}): Promise { + const now = input.now ?? new Date() + const meter = userMeterRpc({ env: input.env, userId: input.userId }) + const { count } = await meter.countRunningPackageServices({ + staleAfterMs: packageServiceStateStaleMs, + excludeService: input.excludeService, + now: now.toISOString(), + }) + return count +} + export async function readEntitlementResourceUsage(input: { db: D1Database userId: string @@ -852,13 +876,13 @@ export async function readEntitlementResourceUsage(input: { `SELECT COUNT(*) AS count FROM jobs WHERE user_id = ?`, [userId], ) - case 'package_services': { - return await countRunningPackageServices({ - db, - userId, - now, - }) - } + case 'package_services': + // Authoritative running liveness is in UserMeter. Callers must use + // readCurrentEntitlementResourceUsage or pass getCurrent with + // countRunningPackageServices to assertWithinEntitlement. + throw new Error( + 'package_services usage must be read from UserMeter (use readCurrentEntitlementResourceUsage or pass getCurrent with countRunningPackageServices).', + ) case 'persistent_package_services': // Finite 0/1 gate: the limit is 0 (not allowed) or 1 (allowed), // so the current count never changes the outcome. @@ -997,6 +1021,13 @@ export async function readCurrentEntitlementResourceUsage(input: { now: input.now, }) } + if (input.resource === 'package_services') { + return await countRunningPackageServices({ + env: input.env, + userId: input.userId, + now: input.now, + }) + } return await readEntitlementResourceUsage({ db: input.db, userId: input.userId, diff --git a/packages/worker/src/entitlements/user-meter-do.ts b/packages/worker/src/entitlements/user-meter-do.ts index c99eca927e..d38bf93b67 100644 --- a/packages/worker/src/entitlements/user-meter-do.ts +++ b/packages/worker/src/entitlements/user-meter-do.ts @@ -27,13 +27,15 @@ export const userMeterDailyCounterRetentionDays = 7 const metaSchemaVersionKey = 'schema_version' /** Bump when initializeSchema DDL changes; warm objects skip DDL. */ const userMeterSchemaVersion = 7 -/** Singleton row id for expand-phase D1 storage-byte shadow (schema v4). */ +/** Singleton row id for authoritative storage-byte state (schema v4). */ const storageBytesStateRowId = 1 /** Singleton row id for deletion fence / write leases (schema v6+). */ const deletionStateRowId = 1 -/** Matches D1 `packageServiceStateStaleMs` (24h); cutover-support RPC only. */ +/** + * Matches D1 `packageServiceStateStaleMs` (24h). Used by the authoritative + * running-count RPC and its D1 parity comparison. + */ export const userMeterPackageServiceStateStaleMs = 24 * 60 * 60 * 1000 - const defaultExportPageSize = 100 const maxExportPageSize = 500 const maxInboundDeliveryIdLength = 256 @@ -294,15 +296,15 @@ export type UserMeterDeletionBootstrapResult = { export type UserMeterExportResult = { counters: Array /** - * Non-authoritative D1 storage-byte shadow. Emitted only on the first - * export page (`startAfter` absent); subsequent pages return `null` so - * paged consumers never double-count the singleton shadow. + * Authoritative storage-byte state, retained under the legacy export field + * name. Emitted only on the first export page (`startAfter` absent); + * subsequent pages return `null` so paged consumers never double-count it. */ storageBytesShadow: UserMeterStorageBytesState | null /** - * Non-authoritative D1 `package_service_states` shadow rows. Emitted only - * on the first export page (`startAfter` absent); subsequent pages return - * `null` so paged consumers never double-count the inventory. + * Authoritative package-service liveness rows, retained under the legacy + * export field name. Emitted only on the first export page (`startAfter` + * absent); subsequent pages return `null`. */ packageServiceStatesShadow: Array | null /** @@ -663,7 +665,7 @@ class UserMeterBase extends DurableObject { `CREATE INDEX IF NOT EXISTS idx_inbound_delivery_claims_day ON inbound_delivery_claims (day)`, ) - // Expand-phase shadow of D1 `users.d1_storage_bytes` (schema v4). + // Authoritative storage-byte state (schema v4). this.ctx.storage.sql.exec(` CREATE TABLE IF NOT EXISTS storage_bytes_state ( id INTEGER PRIMARY KEY NOT NULL CHECK (id = 1), @@ -672,7 +674,7 @@ class UserMeterBase extends DurableObject { updated_at TEXT NOT NULL ) `) - // Expand-phase shadow of D1 `package_service_states` (schema v5). + // Authoritative package-service liveness state (schema v5). this.ctx.storage.sql.exec(` CREATE TABLE IF NOT EXISTS package_service_states ( package_id TEXT NOT NULL, @@ -1065,7 +1067,7 @@ class UserMeterBase extends DurableObject { return readyState(row.count, row.revision) } - /** Cutover-support cold seed; expand-phase shadow sync uses {@link setStorageBytes}. */ + /** Cold seed from the retained D1 mirror when authoritative state is absent. */ async initializeStorageBytes(input: { bytes: number updatedAt: string @@ -1089,7 +1091,7 @@ class UserMeterBase extends DurableObject { return { ...this.storageReadyState(row.bytes, row.revision), created } } - /** Cutover-support / shadow read; expand-phase usage and enforcement use D1. */ + /** Authoritative storage-byte usage read. */ async readStorageBytes(): Promise { const row = this.readStorageRow() if (!row) return { outcome: 'needs_bootstrap' } @@ -1097,8 +1099,7 @@ class UserMeterBase extends DurableObject { } /** - * Cutover-support atomic reserve. Expand-phase enforcement uses D1 only; - * missing state returns `needs_bootstrap`. + * Authoritative atomic reserve. Missing state returns `needs_bootstrap`. */ async reserveStorageBytes(input: { requested: number @@ -1301,8 +1302,8 @@ class UserMeterBase extends DurableObject { } /** - * Expand-phase shadow upsert (monotonic on `sourceUpdatedAt`). Stale writes - * are rejected; expand-phase enforcement still reads D1. + * D1-first lifecycle projection into authoritative liveness state, + * monotonic on `sourceUpdatedAt`. Stale writes are rejected. */ async upsertPackageServiceState(input: { packageId: string @@ -1384,7 +1385,7 @@ class UserMeterBase extends DurableObject { return { applied: true, created: false, state: row } } - /** Expand-phase shadow delete; discovery still uses D1. */ + /** Delete authoritative liveness state; discovery remains in D1. */ async deletePackageServiceState(input: { packageId: string serviceName: string @@ -1400,7 +1401,7 @@ class UserMeterBase extends DurableObject { return { deleted: cursor.rowsWritten > 0 } } - /** Cutover-support paged shadow list; expand-phase discovery uses D1. */ + /** Paged liveness inventory; discovery and account export remain in D1. */ async listPackageServiceStates(input: { pageSize?: number startAfter?: string | null @@ -1451,10 +1452,7 @@ class UserMeterBase extends DurableObject { } } - /** - * Cutover-support running count from the shadow table. Expand-phase - * enforcement still uses D1 `countRunningPackageServices`. - */ + /** Authoritative running count for service-start enforcement. */ async countRunningPackageServices(input: { staleAfterMs?: number excludeService?: { packageId: string; serviceName: string } @@ -1499,7 +1497,7 @@ class UserMeterBase extends DurableObject { return { count: Math.max(0, Number(row?.count ?? 0)) } } - /** Cutover-support bulk seed; same monotonic guard as upsert. */ + /** Explicit migration/repair primitive; same monotonic guard as upsert. */ async bootstrapPackageServiceStates(input: { states: ReadonlyArray<{ packageId: string @@ -2212,20 +2210,20 @@ export type UserMeterRpc = { day: string updatedAt: string }) => Promise - /** Cutover-support cold seed; expand-phase shadow uses setStorageBytes. */ + /** Cold seed from the retained D1 mirror when authoritative state is absent. */ initializeStorageBytes: (input: { bytes: number updatedAt: string }) => Promise - /** Cutover-support / shadow read; expand-phase usage reads D1. */ + /** Authoritative storage-byte usage read. */ readStorageBytes: () => Promise - /** Cutover-support reserve; expand-phase enforcement uses D1. */ + /** Authoritative atomic storage-byte reserve. */ reserveStorageBytes: (input: { requested: number limit: number updatedAt: string }) => Promise - /** Expand-phase absolute shadow-set from D1. */ + /** Absolute maintenance set; use revision CAS for live reconciliation. */ setStorageBytes: (input: { bytes: number updatedAt: string @@ -2240,7 +2238,7 @@ export type UserMeterRpc = { expectedRevision: number updatedAt: string }) => Promise - /** Expand-phase shadow upsert; not used for enforcement. */ + /** Dual-write shadow upsert (D1 writes first); monotonic sourceUpdatedAt guard. */ upsertPackageServiceState: (input: { packageId: string serviceName: string @@ -2249,23 +2247,28 @@ export type UserMeterRpc = { sourceUpdatedAt: string updatedAt?: string }) => Promise - /** Expand-phase shadow delete. */ + /** Shadow delete on service stop/purge. D1 delete runs first. */ deletePackageServiceState: (input: { packageId: string serviceName: string }) => Promise - /** Cutover-support paged shadow list; discovery uses D1. */ + /** Paged inventory list. Discovery and account export still use D1. */ listPackageServiceStates: (input: { pageSize?: number startAfter?: string | null }) => Promise - /** Cutover-support running count; service-start still uses D1. */ + /** Authoritative running count for service-start enforcement. */ countRunningPackageServices: (input: { staleAfterMs?: number excludeService?: { packageId: string; serviceName: string } now?: string }) => Promise - /** Cutover-support bulk seed with sourceUpdatedAt monotonic guards. */ + /** + * Explicit migration/repair primitive: bulk-seed from D1 + * `package_service_states` (all statuses). Uses monotonic `sourceUpdatedAt` + * guard so live dual-write shadow rows are never clobbered. Not on the + * enforcement path — `countRunningPackageServices` reads the DO directly. + */ bootstrapPackageServiceStates: (input: { states: ReadonlyArray<{ packageId: string diff --git a/packages/worker/src/mcp/capabilities/services/service-start.node.test.ts b/packages/worker/src/mcp/capabilities/services/service-start.node.test.ts index eb185de7e4..eee44494af 100644 --- a/packages/worker/src/mcp/capabilities/services/service-start.node.test.ts +++ b/packages/worker/src/mcp/capabilities/services/service-start.node.test.ts @@ -2,6 +2,8 @@ import { expect, test, vi } from 'vitest' import { createMcpCallerContext } from '#mcp/context.ts' import { isEntitlementLimitError } from '#worker/entitlements/errors.ts' import { planLimits } from '#worker/entitlements/plans.ts' +import { createInMemoryUserMeterEnv } from '#worker/test-support/user-meter.ts' +import { type UserMeterRpc } from '#worker/entitlements/user-meter-client.ts' import { createStableUserIdFromEmail } from '#worker/user-id.ts' const mockModule = vi.hoisted(() => ({ @@ -34,10 +36,8 @@ const { serviceStartCapability } = await import('./service-start.ts') function createEntitlementsTestDb(input: { users?: Array<{ email: string; plan: string | null }> - runningServices?: Array<{ packageId: string; name: string }> }) { const users = input.users ?? [] - const runningServices = input.runningServices ?? [] return { prepare(query: string) { @@ -49,20 +49,11 @@ function createEntitlementsTestDb(input: { const user = users.find((row) => row.email === params[0]) return (user ? { plan: user.plan } : null) as T | null } - if (query.includes('FROM package_service_states')) { - const hasExclusion = query.includes('AND NOT (package_id = ?') - const excludedPackageId = hasExclusion ? params[2] : null - const excludedName = hasExclusion ? params[3] : null - const counted = runningServices.filter( - (service) => - !hasExclusion || - service.packageId !== excludedPackageId || - service.name !== excludedName, - ) - return { count: counted.length } as T - } throw new Error(`Unsupported first query: ${query}`) }, + async all(): Promise<{ results: Array }> { + throw new Error(`Unsupported all() query: ${query}`) + }, async run() { throw new Error(`Unsupported run query: ${query}`) }, @@ -162,10 +153,30 @@ function mockServiceRpc(input: { function buildRunningServices(count: number) { return Array.from({ length: count }, (_, index) => ({ packageId: `other-package-${index}`, - name: `service-${index}`, + serviceName: `service-${index}`, })) } +async function seedRunningServicesInMeter( + meterEnv: { USER_METER?: DurableObjectNamespace }, + userId: string, + services: ReadonlyArray<{ packageId: string; serviceName: string }>, +) { + const freshAt = new Date().toISOString() + const meter = meterEnv.USER_METER!.get( + meterEnv.USER_METER!.idFromName(userId), + ) as unknown as UserMeterRpc + for (const service of services) { + await meter.upsertPackageServiceState({ + packageId: service.packageId, + serviceName: service.serviceName, + status: 'running', + startedAt: freshAt, + sourceUpdatedAt: freshAt, + }) + } +} + test('service_start denies persistent services for free plan users', async () => { const email = 'planned@example.com' const userId = await createStableUserIdFromEmail(email) @@ -207,11 +218,15 @@ test('service_start denies bounded starts at the package_services limit', async if (limit === null) { throw new Error('Expected a numeric pro package service limit.') } + const { env: meterEnv } = createInMemoryUserMeterEnv() + await seedRunningServicesInMeter( + meterEnv, + userId, + buildRunningServices(limit), + ) const env = { - APP_DB: createEntitlementsTestDb({ - users: [{ email, plan: 'pro' }], - runningServices: buildRunningServices(limit), - }), + APP_DB: createEntitlementsTestDb({ users: [{ email, plan: 'pro' }] }), + ...meterEnv, } as Env const callerContext = createPlanUserCallerContext({ userId, email }) @@ -246,11 +261,10 @@ test('service_start skips enforcement when the service is already running', asyn if (limit === null) { throw new Error('Expected a numeric pro package service limit.') } + // Enforcement is skipped entirely when the service is already running; + // USER_METER and running service count are not consulted. const env = { - APP_DB: createEntitlementsTestDb({ - users: [{ email, plan: 'pro' }], - runningServices: buildRunningServices(limit), - }), + APP_DB: createEntitlementsTestDb({ users: [{ email, plan: 'pro' }] }), } as Env const callerContext = createPlanUserCallerContext({ userId, email }) @@ -291,14 +305,14 @@ test('service_start lets a stale running row for the same service restart at the // The target service itself holds one of the 'running' telemetry rows // (stale after a hard eviction); excluding it keeps the restart under // the limit. + const { env: meterEnv } = createInMemoryUserMeterEnv() + await seedRunningServicesInMeter(meterEnv, userId, [ + ...buildRunningServices(limit - 1), + { packageId: 'package-123', serviceName: 'realtime-supervisor' }, + ]) const env = { - APP_DB: createEntitlementsTestDb({ - users: [{ email, plan: 'pro' }], - runningServices: [ - ...buildRunningServices(limit - 1), - { packageId: 'package-123', name: 'realtime-supervisor' }, - ], - }), + APP_DB: createEntitlementsTestDb({ users: [{ email, plan: 'pro' }] }), + ...meterEnv, } as Env const callerContext = createPlanUserCallerContext({ userId, email }) @@ -323,13 +337,15 @@ test('service_start allows below-max usage and denies at the max plan ceiling', const email = 'max@example.com' const userId = await createStableUserIdFromEmail(email) const maxLimit = planLimits.max.maxPackageServices + const { env: belowMaxMeterEnv } = createInMemoryUserMeterEnv() + await seedRunningServicesInMeter( + belowMaxMeterEnv, + userId, + buildRunningServices(planLimits.pro.maxPackageServices + 1), + ) const belowMaxEnv = { - APP_DB: createEntitlementsTestDb({ - users: [{ email, plan: 'max' }], - runningServices: buildRunningServices( - planLimits.pro.maxPackageServices + 1, - ), - }), + APP_DB: createEntitlementsTestDb({ users: [{ email, plan: 'max' }] }), + ...belowMaxMeterEnv, } as Env const callerContext = createPlanUserCallerContext({ userId, email }) @@ -349,11 +365,15 @@ test('service_start allows below-max usage and denies at the max plan ceiling', run_id: 'run-123', }) + const { env: atCeilingMeterEnv } = createInMemoryUserMeterEnv() + await seedRunningServicesInMeter( + atCeilingMeterEnv, + userId, + buildRunningServices(maxLimit), + ) const atCeilingEnv = { - APP_DB: createEntitlementsTestDb({ - users: [{ email, plan: 'max' }], - runningServices: buildRunningServices(maxLimit), - }), + APP_DB: createEntitlementsTestDb({ users: [{ email, plan: 'max' }] }), + ...atCeilingMeterEnv, } as Env mockModule.getSavedPackageById.mockResolvedValue( createSavedPackage({ userId }), diff --git a/packages/worker/src/mcp/capabilities/services/service-start.ts b/packages/worker/src/mcp/capabilities/services/service-start.ts index 3d8cdbfc38..ab8a8fa602 100644 --- a/packages/worker/src/mcp/capabilities/services/service-start.ts +++ b/packages/worker/src/mcp/capabilities/services/service-start.ts @@ -74,7 +74,7 @@ export const serviceStartCapability = defineDomainCapability( // restart. getCurrent: async () => await countRunningPackageServices({ - db: ctx.env.APP_DB, + env: ctx.env, userId: serviceContext.user.userId, excludeService: { packageId: serviceContext.savedPackage.id, diff --git a/packages/worker/src/test-support/user-meter.ts b/packages/worker/src/test-support/user-meter.ts index 46a8e066f6..da6afc5999 100644 --- a/packages/worker/src/test-support/user-meter.ts +++ b/packages/worker/src/test-support/user-meter.ts @@ -36,7 +36,7 @@ type DeletionState = { /** * In-memory UserMeter stub keyed by `idFromName` (stable userId) for node-unit - * coverage of expand-phase service/usage paths. Workers suites exercise the + * coverage of authoritative service/usage paths. Workers suites exercise the * real Durable Object binding instead. */ export function createInMemoryUserMeterEnv() { @@ -47,7 +47,6 @@ export function createInMemoryUserMeterEnv() { Map >() const deletionByUser = new Map() - function counterKey(resource: string, day: string) { return `${resource}\0${day}` }