diff --git a/README.md b/README.md index 9988dbb..759980c 100644 --- a/README.md +++ b/README.md @@ -40,6 +40,7 @@ src/ env.ts # zod-validated process.env auth/ # JWT, Google OAuth, email signup users/, accounts/, apiKeys/, pricing/, gateway/, chain/ + lib/ # shared coercion helpers (Decimal → wei string, query params) queue/ # connection, registry, lock, processor base, service t4t/ # VENDORED from t4t-container prisma/schema.prisma diff --git a/package.json b/package.json index c9b2fe5..a9541be 100644 --- a/package.json +++ b/package.json @@ -13,6 +13,7 @@ "db:migrate": "bunx prisma migrate deploy", "db:seed": "bun run prisma/seed/index.ts", "test:e2e": "bun test --preload ./test/setup.ts ./test/**/*.e2e-spec.ts", + "test:unit": "bun test ./test/unit", "typecheck": "bunx tsc --noEmit" }, "//comment": "Prisma 7 reads its config from prisma.config.ts at the package root, not from package.json#prisma.", diff --git a/prisma/migrations/20260805120000_gateway_job_reservation_released_at/migration.sql b/prisma/migrations/20260805120000_gateway_job_reservation_released_at/migration.sql new file mode 100644 index 0000000..2e8e3a1 --- /dev/null +++ b/prisma/migrations/20260805120000_gateway_job_reservation_released_at/migration.sql @@ -0,0 +1,27 @@ +-- AlterTable +ALTER TABLE "GatewayJob" ADD COLUMN "reservationReleasedAt" TIMESTAMP(3); + +-- Data repair: before this migration, `releaseReservation` bailed out whenever +-- the GatewayJob row was already in a terminal status — and both failure paths +-- in `gateway/context.ts` marked the row `cancelled` / `timed_out` *before* +-- calling it. So every job that failed to ACK or to deliver left its +-- `Account.reservedXbzzWei` permanently inflated, shrinking the tenant's +-- available balance for good. Give that stuck xBZZ back, flooring at 0 so a +-- partially-repaired database can never go negative. +UPDATE "Account" a +SET "reservedXbzzWei" = GREATEST(a."reservedXbzzWei" - leaked.total, 0) +FROM ( + SELECT j."userId" AS "userId", SUM(j."estimatedMaxXbzzWei") AS total + FROM "GatewayJob" j + WHERE j."status" IN ('cancelled', 'timed_out') + AND j."reservationReleasedAt" IS NULL + GROUP BY j."userId" +) AS leaked +WHERE a."userId" = leaked."userId"; + +-- Backfill the marker for every job whose reservation is no longer outstanding +-- (either repaired above, or already settled by `settleXbzz` on JobClaimed) so +-- a late JobClaimed event can't decrement the same reservation twice. +UPDATE "GatewayJob" +SET "reservationReleasedAt" = COALESCE("claimedAt", "updatedAt") +WHERE "status" IN ('claimed', 'cancelled', 'timed_out', 'failed'); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 7434f21..8f6a4b9 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -176,6 +176,13 @@ model GatewayJob { ackDeadline DateTime? deliveryDeadline DateTime? + // Set the first time this job's off-chain reservation is released — by + // `settleXbzz` on JobClaimed, or by `releaseReservation` on cancel / timeout + // / dispatch failure. It is the compare-and-set marker that makes the + // release exactly-once: `status` cannot serve that role because a job can be + // marked terminal and released by different code paths in either order. + reservationReleasedAt DateTime? + // Token accounting (from JobClaimed actualPayment) promptTokens Int? completionTokens Int? diff --git a/src/accounts/ledger.ts b/src/accounts/ledger.ts index ff3e1d1..11e4713 100644 --- a/src/accounts/ledger.ts +++ b/src/accounts/ledger.ts @@ -13,6 +13,12 @@ import { logger } from '../logger'; * account), not as ledger rows — the ledger means "money moved" not "money * locked". On `JobClaimed` the reservation is released AND a `settle` ledger * row debits the actual cost. + * + * A reservation must be released exactly once, and the race is real: the + * failure timers, `handleChat`'s catch block and the claim watcher can all + * reach the same job. `GatewayJob.reservationReleasedAt` is the compare-and-set + * marker — whichever transaction flips it from NULL owns the decrement, and + * everyone else no-ops. */ export class InsufficientBalanceError extends Error { @@ -78,9 +84,32 @@ export async function reserveXbzz(userId: string, amountWei: bigint): Promise { + const { count } = await tx.gatewayJob.updateMany({ + where: { id: gatewayJobId, reservationReleasedAt: null }, + data: { reservationReleasedAt: new Date() }, + }); + return count > 0; +} + /** * Settle a completed job: release the reservation and debit the actual cost. - * Idempotent via the `settle:` ledger key. + * The balance debit is idempotent via the `settle:` ledger key; + * the reservation release is idempotent via `reservationReleasedAt`, which also + * covers the case where a failure path already released it (job cancelled + * locally but claimed on-chain anyway). */ export async function settleXbzz( userId: string, @@ -100,11 +129,14 @@ export async function settleXbzz( idempotencyKey: key, }, }); + const releasing = await claimReservationRelease(tx, gatewayJobId); await tx.account.update({ where: { userId }, data: { balanceXbzzWei: { decrement: new Prisma.Decimal(actualXbzzWei.toString()) }, - reservedXbzzWei: { decrement: new Prisma.Decimal(estimateXbzzWei.toString()) }, + ...(releasing + ? { reservedXbzzWei: { decrement: new Prisma.Decimal(estimateXbzzWei.toString()) } } + : {}), }, }); }); @@ -118,10 +150,10 @@ export async function settleXbzz( } /** - * Release a reservation without debiting (job cancelled or timed out). - * Idempotent via `release:` — but we don't write a ledger row, - * we only adjust `reservedXbzzWei` and rely on the GatewayJob status - * transition to mark the release as having happened. + * Release a reservation without debiting (job cancelled, timed out, or never + * dispatched). No ledger row — no money moved, only the lock lifted. Safe to + * call from every failure path and in any order relative to the GatewayJob + * status transition: `reservationReleasedAt` decides who actually decrements. */ export async function releaseReservation( userId: string, @@ -129,14 +161,7 @@ export async function releaseReservation( estimateXbzzWei: bigint, ): Promise { await prisma.$transaction(async (tx) => { - const job = await tx.gatewayJob.findUniqueOrThrow({ - where: { id: gatewayJobId }, - select: { status: true }, - }); - // Only release once — `cancelled` / `timed_out` / `failed` are terminal. - if (job.status === 'cancelled' || job.status === 'timed_out' || job.status === 'failed') { - return; - } + if (!(await claimReservationRelease(tx, gatewayJobId))) return; await tx.account.update({ where: { userId }, data: { reservedXbzzWei: { decrement: new Prisma.Decimal(estimateXbzzWei.toString()) } }, diff --git a/src/accounts/routes.ts b/src/accounts/routes.ts index 3803ce0..897257a 100644 --- a/src/accounts/routes.ts +++ b/src/accounts/routes.ts @@ -2,6 +2,8 @@ import Elysia from 'elysia'; import { prisma } from '../db'; import { jwtGuard } from '../auth/jwt'; import { getUsdEquivalent } from '../pricing/usdDisplay'; +import { weiString, weiToBigInt } from '../lib/decimal'; +import { parseLimit } from '../lib/query'; export const accountsRoutes = new Elysia({ prefix: '/accounts' }) .use(jwtGuard) @@ -18,8 +20,8 @@ export const accountsRoutes = new Elysia({ prefix: '/accounts' }) }, }); - const balanceWei = BigInt(account.balanceXbzzWei.toString()); - const reservedWei = BigInt(account.reservedXbzzWei.toString()); + const balanceWei = weiToBigInt(account.balanceXbzzWei); + const reservedWei = weiToBigInt(account.reservedXbzzWei); const available = balanceWei - reservedWei; const balanceUsd = await getUsdEquivalent(balanceWei).catch(() => null); @@ -28,8 +30,8 @@ export const accountsRoutes = new Elysia({ prefix: '/accounts' }) return { id: account.id, status: account.status, - balanceXbzzWei: account.balanceXbzzWei.toString(), - reservedXbzzWei: account.reservedXbzzWei.toString(), + balanceXbzzWei: weiString(account.balanceXbzzWei), + reservedXbzzWei: weiString(account.reservedXbzzWei), availableXbzzWei: available.toString(), balanceUsd, // cosmetic; null if oracle unavailable availableUsd, @@ -38,7 +40,7 @@ export const accountsRoutes = new Elysia({ prefix: '/accounts' }) }; }) .get('/me/ledger', async ({ user, query }) => { - const limit = Math.min(Number(query.limit ?? 50), 200); + const limit = parseLimit(query.limit); const account = await prisma.account.findUniqueOrThrow({ where: { userId: user.userId }, select: { id: true }, @@ -57,5 +59,5 @@ export const accountsRoutes = new Elysia({ prefix: '/accounts' }) note: true, }, }); - return entries.map((e) => ({ ...e, deltaXbzzWei: e.deltaXbzzWei.toString() })); + return entries.map((e) => ({ ...e, deltaXbzzWei: weiString(e.deltaXbzzWei) })); }); diff --git a/src/chain/claimWatcher.ts b/src/chain/claimWatcher.ts index 73ad7d8..04315ff 100644 --- a/src/chain/claimWatcher.ts +++ b/src/chain/claimWatcher.ts @@ -15,6 +15,7 @@ import { redisConnection } from '../queue/connection'; import { logger } from '../logger'; import { prisma } from '../db'; import { settleXbzz } from '../accounts/ledger'; +import { weiToBigInt } from '../lib/decimal'; import { applyMarkup } from '../pricing/estimate'; import { getGatewayContext } from '../gateway/context'; @@ -88,7 +89,7 @@ async function handleClaimed(jobId: Hex | undefined, paid: bigint | undefined): } if (job.status === 'claimed') return; // idempotent - const estimateWithMarkup = BigInt(job.estimatedMaxXbzzWei.toString()); + const estimateWithMarkup = weiToBigInt(job.estimatedMaxXbzzWei); const actualWithMarkup = applyMarkup(paid); await settleXbzz(job.userId, job.id, estimateWithMarkup, actualWithMarkup); diff --git a/src/gateway/context.ts b/src/gateway/context.ts index 6dc62b0..4cfaa6f 100644 --- a/src/gateway/context.ts +++ b/src/gateway/context.ts @@ -133,19 +133,41 @@ async function bootGatewayContext(): Promise { clearFailureTimers(routing); } + /** Lift the tenant's reservation, then record the terminal status. The order + * no longer matters for correctness (`reservationReleasedAt` is the + * exactly-once marker, not the job status), but a failure to release must + * never stop us from settling the request — otherwise the caller hangs. */ + async function releaseAndMark( + routing: Hex, + meta: JobMetaSlot, + status: 'cancelled' | 'timed_out' | 'failed', + reason: string, + ): Promise { + await releaseReservation(meta.userId, meta.gatewayJobId, meta.estimateMaxXbzzWei).catch((err) => + logger.error({ err, routing, gatewayJobId: meta.gatewayJobId }, 'reservation release failed'), + ); + await markGatewayJob(meta.gatewayJobId, status, reason); + } + async function onAckTimeout(routing: Hex): Promise { const slot = failureTimers.get(routing); const meta = jobMeta.get(routing); if (!slot || !meta) return; + slot.cancelTimer = undefined; try { const tx = await cancelJob(chain, slot.onChainJobId); logger.warn({ tx, routing, onChainJobId: slot.onChainJobId }, 'no ACK — cancelled on-chain'); - await markGatewayJob(meta.gatewayJobId, 'cancelled', 'no PSS ack within ACK_WINDOW'); - await releaseReservation(meta.userId, meta.gatewayJobId, meta.estimateMaxXbzzWei); - rejectAndCleanup(routing, new Error('provider failed to ACK within window')); } catch (err) { - logger.warn({ err, routing }, 'cancelJob failed — provider likely acked mid-flight'); + // Most likely the provider acked mid-flight, so the job is still live and + // cancelling it is no longer allowed. We are done waiting on the ACK + // either way — hand the job to the delivery deadline so *something* still + // settles it. + logger.warn({ err, routing }, 'cancelJob failed — falling through to the delivery deadline'); + armDeliveryTimeout(routing, slot); + return; } + await releaseAndMark(routing, meta, 'cancelled', 'no PSS ack within ACK_WINDOW'); + rejectAndCleanup(routing, new Error('provider failed to ACK within window')); } async function onDeliveryTimeout(routing: Hex): Promise { @@ -155,8 +177,7 @@ async function bootGatewayContext(): Promise { try { const tx = await timeoutJob(chain, slot.onChainJobId); logger.warn({ tx, routing, onChainJobId: slot.onChainJobId }, 'delivery missed — timeoutJob'); - await markGatewayJob(meta.gatewayJobId, 'timed_out', 'no PSS delivery before deadline'); - await releaseReservation(meta.userId, meta.gatewayJobId, meta.estimateMaxXbzzWei); + await releaseAndMark(routing, meta, 'timed_out', 'no PSS delivery before deadline'); rejectAndCleanup(routing, new Error('provider failed to deliver before deadline')); return; } catch (err) { @@ -165,11 +186,14 @@ async function bootGatewayContext(): Promise { try { const tx = await cancelJob(chain, slot.onChainJobId); logger.warn({ tx, routing, onChainJobId: slot.onChainJobId }, 'cancelJob fallback applied'); - await markGatewayJob(meta.gatewayJobId, 'cancelled', 'no PSS delivery; cancelJob fallback'); - await releaseReservation(meta.userId, meta.gatewayJobId, meta.estimateMaxXbzzWei); + await releaseAndMark(routing, meta, 'cancelled', 'no PSS delivery; cancelJob fallback'); rejectAndCleanup(routing, new Error('provider failed to deliver before deadline')); } catch (err) { logger.error({ err, routing }, 'both timeoutJob and cancelJob failed — job stuck in escrow'); + // The escrow entry is stuck, but the tenant shouldn't be: release the + // reservation. If the provider does eventually claim, `settleXbzz` still + // debits the actual cost and skips the (already taken) release. + await releaseAndMark(routing, meta, 'failed', 'delivery deadline missed; settlement failed'); rejectAndCleanup(routing, new Error('delivery deadline missed; settlement failed')); } } @@ -182,50 +206,83 @@ async function bootGatewayContext(): Promise { }, (ACK_WINDOW_SECONDS + FAILURE_GRACE_SECONDS) * 1000); } + /** Arm the delivery-deadline timer once. Every path that stops waiting for + * an ACK must call this, otherwise nothing is left to settle the job and the + * chat request hangs until the client gives up — holding a semaphore slot + * and the tenant's reservation for as long as the process lives. */ + function armDeliveryTimeout(routing: Hex, slot: FailureSlot): void { + if (slot.timeoutTimer) return; + const msUntilTimeout = Math.max( + 0, + (slot.deliveryDeadline + FAILURE_GRACE_SECONDS) * 1000 - Date.now(), + ); + slot.timeoutTimer = setTimeout(() => { + onDeliveryTimeout(routing).catch((err) => + logger.error({ err, routing }, 'onDeliveryTimeout threw'), + ); + }, msUntilTimeout); + } + + /** + * An envelope only counts if it was signed by the provider this gateway + * actually posted the job to. `verifyEnvelope` proves the signature matches + * `envelope.from` — it says nothing about who `from` is, and the routing id + * plus our PSS public key both travel in the (unencrypted) `job_notify` body. + * Without this check anyone who observes a routing id can sign their own + * `job_deliver` and hand the tenant an arbitrary completion. + */ + function envelopeSenderMatches( + envelope: Envelope, + meta: JobMetaSlot | undefined, + ): meta is JobMetaSlot { + if (!meta) return false; + return String(envelope.from).toLowerCase() === meta.provider.toLowerCase(); + } + // Handler for any verified envelope addressed to this gateway, regardless // of which transport (PSS subscription or HTTPS /pss/inbox) brought it. const onEnvelope = async (envelope: Envelope): Promise => { if (envelope.type === 'job_ack') { const body = envelope.body as JobAckBody; - logger.info({ jobId: body.jobId, eta: body.estimatedCompletion }, 'ack received'); const slot = failureTimers.get(body.jobId); const meta = jobMeta.get(body.jobId); + if (!envelopeSenderMatches(envelope, meta)) { + logger.warn( + { jobId: body.jobId, from: envelope.from, type: envelope.type }, + 'dropping envelope — not from the provider this job was posted to', + ); + return; + } + logger.info({ jobId: body.jobId, eta: body.estimatedCompletion }, 'ack received'); if (slot) { if (slot.cancelTimer) { clearTimeout(slot.cancelTimer); slot.cancelTimer = undefined; } - if (!slot.timeoutTimer) { - const msUntilTimeout = Math.max( - 0, - (slot.deliveryDeadline + FAILURE_GRACE_SECONDS) * 1000 - Date.now(), - ); - slot.timeoutTimer = setTimeout(() => { - onDeliveryTimeout(body.jobId).catch((err) => - logger.error({ err, routing: body.jobId }, 'onDeliveryTimeout threw'), - ); - }, msUntilTimeout); - } - } - if (meta) { - await markGatewayJob(meta.gatewayJobId, 'acked', null, { ackedAt: new Date() }); + armDeliveryTimeout(body.jobId, slot); } + await markGatewayJob(meta.gatewayJobId, 'acked', null, { ackedAt: new Date() }); } else if (envelope.type === 'job_deliver') { const body = envelope.body as JobDeliverBody; const slot = pending.get(body.jobId); const meta = jobMeta.get(body.jobId); if (!slot) return; + if (!envelopeSenderMatches(envelope, meta)) { + logger.warn( + { jobId: body.jobId, from: envelope.from, type: envelope.type }, + 'dropping envelope — not from the provider this job was posted to', + ); + return; + } try { const bytes = await downloadChunk({ bee, postageBatchId, logger }, body.responseHash); const payload = await jsonDecrypt(cipher, bytes); - if (meta) { - await markGatewayJob(meta.gatewayJobId, 'delivered', null, { - deliveredAt: new Date(), - responseHash: body.responseHash, - promptTokens: payload.openaiResponse.usage?.prompt_tokens ?? null, - completionTokens: payload.openaiResponse.usage?.completion_tokens ?? null, - }); - } + await markGatewayJob(meta.gatewayJobId, 'delivered', null, { + deliveredAt: new Date(), + responseHash: body.responseHash, + promptTokens: payload.openaiResponse.usage?.prompt_tokens ?? null, + completionTokens: payload.openaiResponse.usage?.completion_tokens ?? null, + }); slot.resolve(payload.openaiResponse); } catch (err) { slot.reject(err); diff --git a/src/gateway/pssInbox.ts b/src/gateway/pssInbox.ts index 0a1656d..4f8fe16 100644 --- a/src/gateway/pssInbox.ts +++ b/src/gateway/pssInbox.ts @@ -11,9 +11,11 @@ import { logger } from '../logger'; * hand off to the shared envelope handler. * * Lives outside the JWT-guarded routes because providers don't have JWTs. - * Security model: anyone can POST, but only envelopes signed by a wallet the - * gateway is actively waiting on (`pending.has(jobId)`) ever produce side - * effects. Unsigned or unmatched envelopes are dropped. + * Security model: anyone can POST, but an envelope only produces side effects + * if (a) its signature recovers to `envelope.from`, checked here, and (b) + * `envelope.from` is the provider the gateway posted that job to, checked in + * `onEnvelope`. Everything else is dropped. The signature check alone is not + * enough — it authenticates the sender, it does not authorize them. */ export const pssInboxRoutes = new Elysia().post( '/pss/inbox', diff --git a/src/gateway/routes.ts b/src/gateway/routes.ts index 32bb3c3..3a1b83b 100644 --- a/src/gateway/routes.ts +++ b/src/gateway/routes.ts @@ -7,6 +7,8 @@ import { Semaphore } from '../queue/semaphore'; import { env } from '../env'; import { logger } from '../logger'; import { InsufficientBalanceError } from '../accounts/ledger'; +import { weiString } from '../lib/decimal'; +import { parseLimit } from '../lib/query'; const semaphore = new Semaphore(env.MAX_CONCURRENT_JOBS); @@ -73,7 +75,10 @@ export const gatewayRoutes = new Elysia() }), ), temperature: t.Optional(t.Number()), - max_tokens: t.Optional(t.Number()), + // Integer-and-positive is enforced here rather than in the pricing + // math: a fractional `max_tokens` makes `BigInt()` throw (500), and a + // negative one budgets a negative reservation. + max_tokens: t.Optional(t.Integer({ minimum: 1 })), stream: t.Optional(t.Boolean()), }, { additionalProperties: true }, @@ -81,7 +86,7 @@ export const gatewayRoutes = new Elysia() }, ) .get('/v1/jobs', async ({ user, query }) => { - const limit = Math.min(Number(query.limit ?? 50), 200); + const limit = parseLimit(query.limit); const jobs = await prisma.gatewayJob.findMany({ where: { userId: user.userId }, orderBy: { createdAt: 'desc' }, @@ -104,7 +109,7 @@ export const gatewayRoutes = new Elysia() }); return jobs.map((j) => ({ ...j, - estimatedMaxXbzzWei: j.estimatedMaxXbzzWei.toString(), - actualXbzzWei: j.actualXbzzWei?.toString() ?? null, + estimatedMaxXbzzWei: weiString(j.estimatedMaxXbzzWei), + actualXbzzWei: j.actualXbzzWei ? weiString(j.actualXbzzWei) : null, })); }); diff --git a/src/lib/decimal.ts b/src/lib/decimal.ts new file mode 100644 index 0000000..dcfe258 --- /dev/null +++ b/src/lib/decimal.ts @@ -0,0 +1,22 @@ +import type { Prisma } from '@prisma/client'; + +/** + * Rendering helpers for the `Decimal(78, 0)` wei columns. + * + * `Prisma.Decimal` is decimal.js under the hood, and decimal.js' `toString()` + * switches to exponential notation once the exponent reaches 21 — so a balance + * of 1000 xBZZ stringifies as `"1e+21"`, not `"1000000000000000000000"`. That + * string is not a valid BigInt literal (`BigInt("1e+21")` throws) and it is not + * something an API client can parse as an integer either. `toFixed()` always + * emits plain digits, which is what wei values need everywhere. + */ + +/** Plain-digit decimal string for a wei column — never exponential notation. */ +export function weiString(value: Prisma.Decimal): string { + return value.toFixed(); +} + +/** Same, as a BigInt. */ +export function weiToBigInt(value: Prisma.Decimal): bigint { + return BigInt(value.toFixed()); +} diff --git a/src/lib/query.ts b/src/lib/query.ts new file mode 100644 index 0000000..d2ab77a --- /dev/null +++ b/src/lib/query.ts @@ -0,0 +1,14 @@ +/** + * Query-string coercion helpers. Everything in `query` is a string (or absent) + * and is attacker-controlled, so it needs clamping before it reaches Prisma — + * `take: NaN` from `?limit=abc` fails the query with a 500 rather than the 400 + * the caller deserves, and a negative `take` silently paginates backwards. + */ +export function parseLimit(raw: unknown, fallback = 50, max = 200): number { + if (raw === undefined || raw === null) return fallback; + // `?limit=` parses as 0, which would clamp to 1 rather than mean "unset". + if (typeof raw === 'string' && raw.trim() === '') return fallback; + const n = Number(raw); + if (!Number.isFinite(n)) return fallback; + return Math.min(Math.max(Math.trunc(n), 1), max); +} diff --git a/src/pricing/estimate.ts b/src/pricing/estimate.ts index 7bba518..058250c 100644 --- a/src/pricing/estimate.ts +++ b/src/pricing/estimate.ts @@ -43,7 +43,15 @@ export function estimateMaxXbzz( req: ChatRequest, offering: Pick, ): { rawWei: bigint; withMarkupWei: bigint } { - const maxTokens = BigInt(req.max_tokens ?? DEFAULT_MAX_TOKENS); + // `max_tokens` reaches us straight off the wire. A fractional value makes + // `BigInt()` throw and a negative one budgets a negative reservation (which + // would *credit* the caller), so anything that isn't a positive integer falls + // back to the default. + const requested = req.max_tokens; + const maxTokens = + typeof requested === 'number' && Number.isInteger(requested) && requested > 0 + ? BigInt(requested) + : DEFAULT_MAX_TOKENS; const budget = maxTokens + HEADROOM_TOKENS; const inPay = BigInt(offering.inputPricePerMillionTokens) * budget; const outPay = BigInt(offering.outputPricePerMillionTokens) * budget; diff --git a/src/queue/service.ts b/src/queue/service.ts index f1ec634..a536bd8 100644 --- a/src/queue/service.ts +++ b/src/queue/service.ts @@ -50,9 +50,13 @@ export function createCudService( }, async delete(args: { where: any }): Promise { + // Snapshot before the delete (the row is gone afterwards), but enqueue + // only once it has actually happened — otherwise a delete that throws + // still tells every consumer the entity is gone. const entity = await prismaDelegate.findUnique({ where: args.where }); - await queue.add('deleted', { [queueName]: serializeBigInts(entity) }); - return prismaDelegate.delete(args); + const deleted = await prismaDelegate.delete(args); + await queue.add('deleted', { [queueName]: serializeBigInts(entity ?? deleted) }); + return deleted; }, }; } diff --git a/test/unit/wei-and-limits.spec.ts b/test/unit/wei-and-limits.spec.ts new file mode 100644 index 0000000..0200a68 --- /dev/null +++ b/test/unit/wei-and-limits.spec.ts @@ -0,0 +1,94 @@ +/** + * Pure-function regressions for the wei/query coercion helpers. No Postgres, + * Redis, Bee or RPC — run with `bun run test:unit`. + */ +import { describe, expect, it } from 'bun:test'; +import { Prisma } from '@prisma/client'; +import { weiString, weiToBigInt } from '../../src/lib/decimal'; +import { parseLimit } from '../../src/lib/query'; + +// `src/pricing/estimate` pulls in the zod-validated env module, which exits the +// process when required vars are missing. Fill in the minimum before importing. +process.env.DATABASE_URL ??= 'postgresql://user:pass@localhost:5432/test'; +process.env.JWT_SECRET_KEY ??= 'test-secret-at-least-16-chars'; +process.env.OPERATOR_PRIVATE_KEY ??= '0x' + '11'.repeat(32); +process.env.GNOSIS_RPC_URL ??= 'http://localhost:8545'; +process.env.REGISTRY_ADDRESS ??= '0x' + '22'.repeat(20); +process.env.ESCROW_ADDRESS ??= '0x' + '33'.repeat(20); +process.env.XBZZ_ADDRESS ??= '0x' + '44'.repeat(20); +process.env.BEE_API_URL ??= 'http://localhost:1633'; +process.env.MARKUP_BPS ??= '2000'; +const { estimateMaxXbzz } = await import('../../src/pricing/estimate'); + +const OFFERING = { + inputPricePerMillionTokens: 10n ** 18n, + outputPricePerMillionTokens: 2n * 10n ** 18n, + maxContextTokens: 0n, +}; + +describe('wei rendering', () => { + it('keeps large balances in plain notation', () => { + // decimal.js' toString() flips to exponential at 1e21 — i.e. at 1000 xBZZ, + // which is an ordinary balance. "1e+21" is not a valid BigInt literal. + const thousandXbzz = new Prisma.Decimal('1000000000000000000000'); + expect(thousandXbzz.toString()).toBe('1e+21'); // the trap being guarded + expect(weiString(thousandXbzz)).toBe('1000000000000000000000'); + expect(weiToBigInt(thousandXbzz)).toBe(1000000000000000000000n); + }); + + it('round-trips a full uint256', () => { + const max = (2n ** 256n - 1n).toString(); + expect(weiToBigInt(new Prisma.Decimal(max))).toBe(2n ** 256n - 1n); + }); + + it('handles negative deltas and zero', () => { + expect(weiString(new Prisma.Decimal('-5000000000000000000000'))).toBe('-5000000000000000000000'); + expect(weiString(new Prisma.Decimal(0))).toBe('0'); + }); +}); + +describe('parseLimit', () => { + it('clamps to the allowed range', () => { + expect(parseLimit('10')).toBe(10); + expect(parseLimit('1000')).toBe(200); + expect(parseLimit('-5')).toBe(1); + expect(parseLimit('0')).toBe(1); + }); + + it('falls back instead of handing Prisma a NaN take', () => { + expect(parseLimit('abc')).toBe(50); + expect(parseLimit(undefined)).toBe(50); + expect(parseLimit('')).toBe(50); // Number('') is 0, but empty means "unset" + }); + + it('truncates fractional input', () => { + expect(parseLimit('12.9')).toBe(12); + }); +}); + +describe('estimateMaxXbzz', () => { + it('never returns a negative reservation', () => { + for (const max_tokens of [-1000, -1, 0]) { + const { rawWei, withMarkupWei } = estimateMaxXbzz({ model: 'm', max_tokens }, OFFERING); + expect(rawWei > 0n).toBe(true); + expect(withMarkupWei >= rawWei).toBe(true); + } + }); + + it('does not throw on a fractional max_tokens', () => { + const { rawWei } = estimateMaxXbzz({ model: 'm', max_tokens: 100.5 }, OFFERING); + const { rawWei: dflt } = estimateMaxXbzz({ model: 'm' }, OFFERING); + expect(rawWei).toBe(dflt); + }); + + it('scales with max_tokens', () => { + const small = estimateMaxXbzz({ model: 'm', max_tokens: 1 }, OFFERING).rawWei; + const large = estimateMaxXbzz({ model: 'm', max_tokens: 100_000 }, OFFERING).rawWei; + expect(large > small).toBe(true); + }); + + it('applies the markup on top of the on-chain amount', () => { + const { rawWei, withMarkupWei } = estimateMaxXbzz({ model: 'm', max_tokens: 1024 }, OFFERING); + expect(withMarkupWei).toBe((rawWei * 12_000n) / 10_000n); + }); +});