diff --git a/apps/gateway-worker/README.md b/apps/gateway-worker/README.md index 5b504999..85908097 100644 --- a/apps/gateway-worker/README.md +++ b/apps/gateway-worker/README.md @@ -106,6 +106,10 @@ reconciles dormant objects to that shape when they are next activated. Run creation is also durably idempotent in Postgres, so request-cache evolution cannot create a duplicate run. +`QuotaTracker` has no compatibility migration path. New objects initialize +directly into the current exact schema, while existing objects must already +match that schema before any quota operation is admitted. + `/v1/tools` and `/v1/agents` read the shared framework-free capability catalog from `@cheatcode/types`. The Mastra registries are statically constrained to the same exact names; workflows are exposed through tools and are not reported as diff --git a/apps/gateway-worker/src/durable-objects/quota-tracker-storage.ts b/apps/gateway-worker/src/durable-objects/quota-tracker-storage.ts index 37598334..195f5a6a 100644 --- a/apps/gateway-worker/src/durable-objects/quota-tracker-storage.ts +++ b/apps/gateway-worker/src/durable-objects/quota-tracker-storage.ts @@ -1,11 +1,8 @@ import { assertExactSqliteSchema, - assertSqliteRowCountPreserved, type ExpectedSqliteObject, - reconcileExactSqliteStorage, setCurrentSqliteStorageVersion, } from "@cheatcode/durable-storage"; -import { QUOTA_FEATURES } from "@cheatcode/types/quota"; const MAX_FINITE_REAL = "1.7976931348623157e308"; const FEATURE_CHECK = "feature IN ('composio_calls', 'sandbox_hours')"; @@ -46,8 +43,6 @@ const USAGE_EVENT_INDEX_SQL = "CREATE INDEX usage_event_feature_time_idx ON usage_event(feature, recorded_at)"; const QUOTA_OPERATION_INDEX_SQL = "CREATE INDEX quota_operation_feature_time_idx ON quota_operation(feature, recorded_at)"; -const QUOTA_TABLES = ["counter", "limit_override", "usage_event", "quota_operation"] as const; -const CURRENT_QUOTA_FEATURES = [QUOTA_FEATURES.composioCalls, QUOTA_FEATURES.sandboxHours] as const; const QUOTA_STORAGE_SCHEMA: readonly ExpectedSqliteObject[] = [ { name: "counter", sql: COUNTER_SQL, tableName: "counter", type: "table" }, @@ -78,49 +73,13 @@ const QUOTA_STORAGE_SCHEMA: readonly ExpectedSqliteObject[] = [ }, ]; -/** - * Opens quota storage and transactionally upgrades an older supported schema. - * - * Durable Object input gates serialize this synchronous admission path. The - * guarded maintenance route still calls the same reconciler for planned bulk - * verification, but ordinary forward-compatible releases cannot strand a - * dormant user object on its previous schema. - */ +/** Opens quota storage only when it matches the exact current contract. */ export function ensureQuotaTrackerStorage(ctx: DurableObjectState): void { if (!hasQuotaTrackerStorage(ctx)) { initializeQuotaTrackerStorage(ctx); return; } - reconcileExactSqliteStorage( - "reconcile", - () => assertQuotaTrackerStorage(ctx), - () => reconcileQuotaTrackerStorage(ctx), - ); -} - -/** - * Force-normalizes quota storage while preserving every current product row. - * - * Quotas removed from the product are intentionally discarded before the - * exact-schema rebuild. Feature values must still be valid text so corrupt - * rows cannot be mistaken for safely retired data. - */ -export function reconcileQuotaTrackerStorage(ctx: DurableObjectState): void { - ctx.storage.transactionSync(() => { - ensureSourceTables(ctx); - const limitColumns = ctx.storage.sql.exec("PRAGMA table_info(limit_override)").toArray(); - if (!hasColumn(limitColumns, "feature") || !hasColumn(limitColumns, "limit_val")) { - throw new Error("Unsupported quota limit schema; refusing lossy evolution."); - } - const hasEntitlementVersion = hasColumn(limitColumns, "entitlement_version"); - assertQuotaFeatureTypes(ctx); - deleteRetiredQuotaRows(ctx); - assertQuotaSourceRows(ctx, hasEntitlementVersion); - const entitlementVersion = hasEntitlementVersion ? "entitlement_version" : "0"; - rebuildQuotaTables(ctx, entitlementVersion); - setCurrentSqliteStorageVersion(ctx); - assertQuotaTrackerStorage(ctx); - }); + assertQuotaTrackerStorage(ctx); } export function assertQuotaTrackerStorage(ctx: DurableObjectState): void { @@ -128,13 +87,19 @@ export function assertQuotaTrackerStorage(ctx: DurableObjectState): void { } function initializeQuotaTrackerStorage(ctx: DurableObjectState): void { - ensureSourceTables(ctx); - ctx.storage.sql.exec(USAGE_EVENT_INDEX_SQL.replace("CREATE INDEX", "CREATE INDEX IF NOT EXISTS")); - ctx.storage.sql.exec( - QUOTA_OPERATION_INDEX_SQL.replace("CREATE INDEX", "CREATE INDEX IF NOT EXISTS"), - ); - setCurrentSqliteStorageVersion(ctx); - assertQuotaTrackerStorage(ctx); + ctx.storage.transactionSync(() => { + for (const sql of [COUNTER_SQL, LIMIT_OVERRIDE_SQL, USAGE_EVENT_SQL, QUOTA_OPERATION_SQL]) { + ctx.storage.sql.exec(sql.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS")); + } + ctx.storage.sql.exec( + USAGE_EVENT_INDEX_SQL.replace("CREATE INDEX", "CREATE INDEX IF NOT EXISTS"), + ); + ctx.storage.sql.exec( + QUOTA_OPERATION_INDEX_SQL.replace("CREATE INDEX", "CREATE INDEX IF NOT EXISTS"), + ); + setCurrentSqliteStorageVersion(ctx); + assertQuotaTrackerStorage(ctx); + }); } export function hasQuotaTrackerStorage(ctx: DurableObjectState): boolean { @@ -146,140 +111,3 @@ export function hasQuotaTrackerStorage(ctx: DurableObjectState): boolean { .toArray().length > 0 ); } - -function ensureSourceTables(ctx: DurableObjectState): void { - for (const sql of [COUNTER_SQL, LIMIT_OVERRIDE_SQL, USAGE_EVENT_SQL, QUOTA_OPERATION_SQL]) { - ctx.storage.sql.exec(sql.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS")); - } -} - -function rebuildQuotaTables(ctx: DurableObjectState, entitlementVersion: string): void { - ctx.storage.sql.exec("DROP INDEX IF EXISTS usage_event_feature_time_idx"); - ctx.storage.sql.exec("DROP INDEX IF EXISTS quota_operation_feature_time_idx"); - for (const table of QUOTA_TABLES) { - ctx.storage.sql.exec(`ALTER TABLE ${table} RENAME TO ${table}_reconcile_source`); - } - for (const sql of [COUNTER_SQL, LIMIT_OVERRIDE_SQL, USAGE_EVENT_SQL, QUOTA_OPERATION_SQL]) { - ctx.storage.sql.exec(sql); - } - copyQuotaRows(ctx, entitlementVersion); - for (const table of QUOTA_TABLES) { - assertSqliteRowCountPreserved(ctx, `${table}_reconcile_source`, table); - ctx.storage.sql.exec(`DROP TABLE ${table}_reconcile_source`); - } - ctx.storage.sql.exec(USAGE_EVENT_INDEX_SQL); - ctx.storage.sql.exec(QUOTA_OPERATION_INDEX_SQL); -} - -function copyQuotaRows(ctx: DurableObjectState, entitlementVersion: string): void { - ctx.storage.sql.exec( - `INSERT INTO counter (feature, period_key, used, updated_at) - SELECT feature, period_key, used, updated_at FROM counter_reconcile_source`, - ); - ctx.storage.sql.exec( - `INSERT INTO limit_override (feature, limit_val, entitlement_version) - SELECT feature, limit_val, ${entitlementVersion} FROM limit_override_reconcile_source`, - ); - ctx.storage.sql.exec( - `INSERT INTO usage_event (id, feature, amount, recorded_at) - SELECT id, feature, amount, recorded_at FROM usage_event_reconcile_source`, - ); - ctx.storage.sql.exec( - `INSERT INTO quota_operation - (event_id, operation, feature, period_key, amount, allowed, - limit_val, remaining, used, recorded_at) - SELECT event_id, operation, feature, period_key, amount, allowed, - limit_val, remaining, used, recorded_at - FROM quota_operation_reconcile_source`, - ); -} - -function assertQuotaFeatureTypes(ctx: DurableObjectState): void { - for (const table of QUOTA_TABLES) { - if ( - ctx.storage.sql - .exec(`SELECT 1 FROM ${table} WHERE typeof(feature) <> 'text' LIMIT 1`) - .toArray().length > 0 - ) { - throw new Error(`Quota ${table} contains a malformed feature; refusing lossy evolution.`); - } - } -} - -function deleteRetiredQuotaRows(ctx: DurableObjectState): void { - for (const table of QUOTA_TABLES) { - ctx.storage.sql.exec( - `DELETE FROM ${table} WHERE feature NOT IN (?, ?)`, - ...CURRENT_QUOTA_FEATURES, - ); - } -} - -function assertQuotaSourceRows(ctx: DurableObjectState, hasEntitlementVersion: boolean): void { - assertNoInvalidRows( - ctx, - "counter", - `SELECT 1 FROM counter WHERE - typeof(feature) <> 'text' OR feature NOT IN (?, ?) OR - typeof(period_key) <> 'text' OR length(period_key) <> 7 OR - period_key NOT GLOB '[0-9][0-9][0-9][0-9]-[0-9][0-9]' OR - typeof(used) NOT IN ('integer', 'real') OR used < 0 OR abs(used) > ${MAX_FINITE_REAL} OR - typeof(updated_at) <> 'integer' OR updated_at < 0 LIMIT 1`, - ); - const entitlementPredicate = hasEntitlementVersion - ? " OR typeof(entitlement_version) <> 'integer' OR entitlement_version < 0" - : ""; - assertNoInvalidRows( - ctx, - "limit_override", - `SELECT 1 FROM limit_override WHERE - typeof(feature) <> 'text' OR feature NOT IN (?, ?) OR - typeof(limit_val) NOT IN ('integer', 'real') OR limit_val < 0 OR - abs(limit_val) > ${MAX_FINITE_REAL}${entitlementPredicate} LIMIT 1`, - ); - assertNoInvalidRows( - ctx, - "usage_event", - `SELECT 1 FROM usage_event WHERE - typeof(id) <> 'integer' OR typeof(feature) <> 'text' OR feature NOT IN (?, ?) OR - typeof(amount) NOT IN ('integer', 'real') OR amount <= 0 OR - abs(amount) > ${MAX_FINITE_REAL} OR typeof(recorded_at) <> 'integer' OR - recorded_at < 0 LIMIT 1`, - ); - assertNoInvalidRows( - ctx, - "quota_operation", - `SELECT 1 FROM quota_operation WHERE - typeof(event_id) <> 'text' OR length(event_id) NOT BETWEEN 1 AND 200 OR - typeof(operation) <> 'text' OR operation NOT IN ('record', 'try-consume') OR - typeof(feature) <> 'text' OR feature NOT IN (?, ?) OR - typeof(period_key) <> 'text' OR length(period_key) <> 7 OR - period_key NOT GLOB '[0-9][0-9][0-9][0-9]-[0-9][0-9]' OR - typeof(amount) NOT IN ('integer', 'real') OR amount <= 0 OR - abs(amount) > ${MAX_FINITE_REAL} OR typeof(allowed) <> 'integer' OR - allowed NOT IN (0, 1) OR typeof(limit_val) NOT IN ('integer', 'real') OR - limit_val < 0 OR abs(limit_val) > ${MAX_FINITE_REAL} OR - typeof(remaining) NOT IN ('integer', 'real') OR remaining < 0 OR - abs(remaining) > ${MAX_FINITE_REAL} OR typeof(used) NOT IN ('integer', 'real') OR - used < 0 OR abs(used) > ${MAX_FINITE_REAL} OR - typeof(recorded_at) <> 'integer' OR recorded_at < 0 LIMIT 1`, - ); -} - -function assertNoInvalidRows( - ctx: DurableObjectState, - table: (typeof QUOTA_TABLES)[number], - sql: string, -): void { - if (ctx.storage.sql.exec(sql, ...CURRENT_QUOTA_FEATURES).toArray().length > 0) { - throw new Error(`Quota ${table} contains invalid current data; refusing lossy evolution.`); - } -} - -function hasColumn(rows: unknown[], name: string): boolean { - return rows.some((row) => isRecord(row) && row["name"] === name); -} - -function isRecord(value: unknown): value is Record { - return typeof value === "object" && value !== null; -} diff --git a/apps/gateway-worker/src/durable-objects/quota-tracker.ts b/apps/gateway-worker/src/durable-objects/quota-tracker.ts index b3fccceb..93463a63 100644 --- a/apps/gateway-worker/src/durable-objects/quota-tracker.ts +++ b/apps/gateway-worker/src/durable-objects/quota-tracker.ts @@ -1,7 +1,6 @@ import { DurableObject } from "cloudflare:workers"; import { assertStorageReconciliationRequest, - reconcileExactSqliteStorage, storageSchemaEvidence, } from "@cheatcode/durable-storage"; import { readJsonRequest } from "@cheatcode/observability"; @@ -35,7 +34,6 @@ import { assertQuotaTrackerStorage, ensureQuotaTrackerStorage, hasQuotaTrackerStorage, - reconcileQuotaTrackerStorage, } from "./quota-tracker-storage"; import { assertGatewayDurableObjectOpen, @@ -131,11 +129,7 @@ export class QuotaTracker extends DurableObject { value: InternalDurableObjectStorageRequest, ): InternalDurableObjectStorageResponse { const input = assertStorageReconciliationRequest(this.ctx, this.env, value, "QuotaTracker"); - reconcileExactSqliteStorage( - input.mode, - () => assertQuotaTrackerStorage(this.ctx), - () => reconcileQuotaTrackerStorage(this.ctx), - ); + assertQuotaTrackerStorage(this.ctx); this.isStorageInitialized = true; return storageSchemaEvidence(input); }