fix(app-hosting): at-most-once tail settle, stop/meter watermark discipline, payer alignment, wake seam (#2502/#2493/#2491 follow-up)

packages/lib/src/monitoring/__tests__/ai-monitoring.test.ts

@@ -720,6 +720,38 @@ describe('trackAIUsage', () => {
     );
   });
 
+  it('should release the hold when writeAiUsage THROWS, not just when it resolves null', async () => {
+    // The gate's reservation would otherwise sit against the payer's spendable
+    // balance for its whole TTL — a throw here never confirmed a charge, so
+    // there is nothing this release could double-free.
+    mockWriteAiUsage.mockRejectedValueOnce(new Error('db error'));
+    const outcome = await trackAIUsage({
+      userId: 'user-1',
+      provider: 'openai',
+      model: 'gpt-4o',
+      holdId: 'hold-thrown',
+    });
+    expect(outcome).toEqual({ persisted: false, creditsSettled: false });
+    expect(mockReleaseHold).toHaveBeenCalledWith('hold-thrown');
+  });
+
+  it('should release the hold when the usage CALCULATION itself throws, before writeAiUsage is even reached', async () => {
+    // A getter that throws on `model` forces the OUTER catch — `calculateCost`
+    // reads it before `writeAiUsage` is ever called, so this exercises the
+    // catch that never even attempted a write.
+    const data: Record<string, unknown> = { userId: 'user-1', provider: 'openai', holdId: 'hold-calc-thrown' };
+    Object.defineProperty(data, 'model', {
+      get() {
+        throw new Error('boom');
+      },
+    });
+
+    const outcome = await trackAIUsage(data as never);
+
+    expect(outcome).toEqual({ persisted: false, creditsSettled: false });
+    expect(mockReleaseHold).toHaveBeenCalledWith('hold-calc-thrown');
+  });
+
   it('given AI request completes, should NOT write prompt or completion content to ai_usage_logs (#957 — GDPR data minimization)', async () => {
     await trackAIUsage({
       userId: 'user-1',

packages/lib/src/monitoring/ai-monitoring.ts

@@ -1169,12 +1169,21 @@ export async function trackAIUsage(data: AIUsageData): Promise<UsageTrackingOutc
         source: data.source,
         holdId: data.holdId,
       });
+      // The reservation would otherwise sit against the payer's spendable balance
+      // until its TTL. Nothing above committed a charge against it on this path —
+      // the throw means `writeAiUsage`/`consumeCredits` never confirmed one — so
+      // there is nothing this release could double-free.
+      if (data.holdId) await releaseHold(data.holdId);
       return USAGE_TRACKING_LOST;
     }
   } catch (error) {
-    loggers.ai.debug('AI usage calculation failed', { 
-      error: (error as Error).message 
+    loggers.ai.debug('AI usage calculation failed', {
+      error: (error as Error).message
     });
+    // Same reasoning as the inner catch above: a stranded hold outlives its TTL
+    // suppressing the payer's balance for nothing, and nothing on this path could
+    // have confirmed a charge against it.
+    if (data.holdId) await releaseHold(data.holdId);
     return USAGE_TRACKING_LOST;
   }
 }

packages/lib/src/services/app-hosting/__tests__/app-lifecycle-metering.test.ts

@@ -180,6 +180,23 @@ describe('wakePublishedApp', () => {
     expect(startMachine).not.toHaveBeenCalled();
   });
 
+  // The full dedicated-tier wake contract (no gate, no payer resolution, no
+  // billing watermark) is covered below in "does NOT gate a dedicated wake",
+  // "opens NO billing window for a dedicated wake", and "does not even resolve
+  // a payer for a dedicated wake". This one covers the case those don't: a
+  // start failure has nothing to release, because a dedicated wake never
+  // placed a hold in the first place.
+  it('given a DEDICATED app whose machine fails to start, should release NOTHING — there was no hold to release', async () => {
+    const { deps, releaseHold, startMachine } = makeDeps();
+    startMachine.mockRejectedValue(new Error('capacity'));
+    seed(appRow({ tier: 'dedicated' }));
+
+    const result = await wakePublishedApp('app-1', deps);
+
+    expect(result).toEqual({ outcome: 'start_failed', error: 'capacity' });
+    expect(releaseHold).not.toHaveBeenCalled();
+  });
+
   it('opens the awake window at the wake instant and records the boundary', async () => {
     const { deps } = makeDeps();
     const row = appRow();
@@ -326,7 +343,8 @@ describe('wakePublishedApp — the abandoned tail a failed close left behind', (
 
   it('bills the stranded span at the boundary the window REALLY ended on, not up to now', async () => {
     const { deps, trackUsage } = makeDeps();
-    seed(abandoned(), [[appRow({ status: 'running' })]]);
+    // Two returning-row entries: the tail's own CAS claim, then the wake's final CAS.
+    seed(abandoned(), [[{ id: 'app-1' }], [appRow({ status: 'running' })]]);
 
     await wakePublishedApp('app-1', deps);
 
@@ -366,7 +384,7 @@ describe('wakePublishedApp — the abandoned tail a failed close left behind', (
         lastStopAt: new Date('2026-08-20T10:10:00.000Z'), // zero-length window
         awakeHoldId: 'hold-stranded',
       }),
-      [[appRow({ status: 'running' })]],
+      [[{ id: 'app-1' }], [appRow({ status: 'running' })]],
     );
 
     await wakePublishedApp('app-1', deps);
@@ -380,7 +398,7 @@ describe('wakePublishedApp — the abandoned tail a failed close left behind', (
     // point — two independent failures at two separate moments — and said so.
     const { deps, trackUsage, startMachine } = makeDeps();
     trackUsage.mockRejectedValue(new Error('ledger down'));
-    seed(abandoned(), [[appRow({ status: 'running' })]]);
+    seed(abandoned(), [[{ id: 'app-1' }], [appRow({ status: 'running' })]]);
 
     const result = await wakePublishedApp('app-1', deps);
 
@@ -397,6 +415,55 @@ describe('wakePublishedApp — the abandoned tail a failed close left behind', (
 
     expect(trackUsage).not.toHaveBeenCalled();
   });
+
+  it('claims the tail EXCLUSIVELY before charging: a wake that starts, fails, and retries does NOT re-bill it', async () => {
+    // `settleAbandonedTail` runs on EVERY wake attempt against a non-running row —
+    // including a retry of a wake whose `startMachine` just failed. Without an
+    // at-most-once claim, each retry would re-settle the same stranded span,
+    // writing a new `ai_usage_logs` row every time.
+    const { deps, trackUsage, startMachine } = makeDeps();
+    startMachine.mockRejectedValueOnce(new Error('fly outage'));
+    const row = abandoned();
+    seed(row, [[{ id: 'app-1' }]]); // the tail claim succeeds; nothing else needed — start fails first
+
+    const first = await wakePublishedApp('app-1', deps);
+    expect(first.outcome).toBe('start_failed');
+    expect(trackUsage).toHaveBeenCalledTimes(1);
+
+    // Re-seed exactly as the CAS would have left the row: watermark and hold
+    // cleared by the first attempt's claim, regardless of the start failure.
+    seed(appRow({ status: 'stopped', driveId: row.driveId, awakeBilledThrough: null, lastStopAt: row.lastStopAt, awakeHoldId: null }), [
+      [appRow({ status: 'running' })],
+    ]);
+
+    const second = await wakePublishedApp('app-1', deps);
+
+    assert({
+      given: 'a wake retried after `start_failed` following a settled tail',
+      should: 'start cleanly and NOT re-bill the already-claimed tail',
+      actual: { outcome: second.outcome, trackUsageCalls: trackUsage.mock.calls.length },
+      expected: { outcome: 'woken', trackUsageCalls: 1 },
+    });
+  });
+
+  it('two wakes racing the same abandoned tail bill it EXACTLY ONCE', async () => {
+    const { deps, trackUsage } = makeDeps();
+    const row = abandoned();
+    // Wake #1 wins the claim (returns the row) and its own final CAS. Wake #2 —
+    // reading the identical stale snapshot, as a genuine race would — loses the
+    // claim (returns no row) and must never reach `trackUsage`.
+    seed(row, [
+      [{ id: 'app-1' }],
+      [appRow({ status: 'running' })],
+      [],
+      [appRow({ status: 'running' })],
+    ]);
+
+    await wakePublishedApp('app-1', deps);
+    await wakePublishedApp('app-1', deps);
+
+    expect(trackUsage).toHaveBeenCalledTimes(1);
+  });
 });
 
 describe('stopPublishedApp', () => {
@@ -584,6 +651,12 @@ describe('stopPublishedApp', () => {
     expect(releaseHold).toHaveBeenCalledWith('hold-1');
   });
 
+  // A meter-tick-during-the-Fly-call race is now closed by `stopPublishedApp`
+  // serializing its ENTIRE sequence (read, Fly call, settle) under the awake
+  // meter's own advisory lock, rather than by re-reading the watermark after the
+  // Fly call — see the "the awake meter's advisory lock" describe block below,
+  // which covers exactly this interleaving.
+
   it('refuses to stop a row that is not running', async () => {
     const { deps } = makeDeps();
     seed(appRow({ status: 'stopped' }));

packages/lib/src/services/app-hosting/__tests__/router.test.ts

@@ -6,8 +6,8 @@
  * could be lost without the pure test noticing:
  *
  *   • the kill switch short-circuits BEFORE any database read;
- *   • the balance is asked about the row's OWN payer (`ownerId`), the same
- *     column the awake-seconds meter charges;
+ *   • the balance is asked about the app's DRIVE-OWNER payer, resolved the SAME
+ *     way the awake-seconds meter resolves it — never `published_apps.ownerId`;
  *   • the ledger is not consulted at all for a dedicated app;
  *   • a router that cannot derive a replay key refuses rather than emitting a
  *     replay with a blank state;
@@ -23,7 +23,7 @@ vi.mock('@pagespace/db/schema/published-apps', () => ({
     status: 'status',
     tier: 'tier',
     machineId: 'machineId',
-    ownerId: 'ownerId',
+    driveId: 'driveId',
     subdomain: 'subdomain',
   },
 }));
@@ -34,6 +34,11 @@ vi.mock('@pagespace/db/operators', () => ({ eq: (a: unknown, b: unknown) => ({ e
 // covered in `billing/__tests__/credit-gate.test.ts`.
 vi.mock('../../../billing/credit-gate', () => ({ hasSpendableBalance: vi.fn() }));
 vi.mock('../../../billing/credit-balance', () => ({ resolveTier: vi.fn() }));
+// Same reasoning: `app-billing`'s default payer resolver drags in the whole
+// credit/monitoring stack to build. Asserted BEHAVIOURALLY (does the router's
+// default binding delegate to it with the right args), which needs a mock, not
+// the real module.
+vi.mock('../app-billing', () => ({ defaultAppBillingDeps: { resolvePayerId: vi.fn() } }));
 
 import {
   defaultAppRouterDeps,
@@ -45,6 +50,7 @@ import { db } from '@pagespace/db/db';
 import { publishedApps } from '@pagespace/db/schema/published-apps';
 import { hasSpendableBalance } from '../../../billing/credit-gate';
 import { resolveTier } from '../../../billing/credit-balance';
+import { defaultAppBillingDeps } from '../app-billing';
 import { isAppHostingEnabled } from '../app-hosting-env';
 import { resolveAppReplaySecret, resolvePublishedAppsApex } from '../routing-env';
 
@@ -57,7 +63,7 @@ function row(overrides: Partial<PublishedAppRouteRow> = {}): PublishedAppRouteRo
     status: 'running',
     tier: 'metered',
     machineId: 'm-1',
-    ownerId: 'user_payer',
+    driveId: 'drive_payer',
     ...overrides,
   };
 }
@@ -68,6 +74,7 @@ function deps(overrides: Partial<AppRouterDeps> = {}): AppRouterDeps {
     apex: () => 'pagespace.app',
     replaySecret: () => SECRET,
     findAppBySubdomain: async () => row(),
+    resolvePayerId: async ({ driveId }) => (driveId === 'drive_payer' ? 'user_payer' : null),
     resolveTier: async () => 'pro',
     hasSpendableBalance: async () => true,
     stampHit: async () => {},
@@ -118,22 +125,39 @@ describe('resolveAppRoute — hostname resolution', () => {
   });
 });
 
-describe('resolveAppRoute — the balance is asked about the row own payer', () => {
-  it("given a metered app, should ask about the row's ownerId, not any other user", async () => {
+describe('resolveAppRoute — the balance is asked about the SAME payer the meter charges', () => {
+  it("given a metered app, should resolve the payer from the row's driveId, not published_apps.ownerId", async () => {
+    const resolvePayerId = vi.fn(async ({ driveId }: { driveId: string }) =>
+      driveId === 'drive_xyz' ? 'user_drive_owner' : null,
+    );
     const hasSpendableBalance = vi.fn(async () => true);
     const resolveTier = vi.fn(async () => 'pro');
     await resolveAppRoute(
       'acme.pagespace.app',
       deps({
-        findAppBySubdomain: async () => row({ ownerId: 'user_drive_owner' }),
+        findAppBySubdomain: async () => row({ driveId: 'drive_xyz' }),
+        resolvePayerId,
         resolveTier,
         hasSpendableBalance,
       }),
     );
+    expect(resolvePayerId).toHaveBeenCalledWith({ driveId: 'drive_xyz' });
     expect(resolveTier).toHaveBeenCalledWith('user_drive_owner');
     expect(hasSpendableBalance).toHaveBeenCalledWith('user_drive_owner', 'pro');
   });
 
+  it('given an unresolvable drive, should refuse (fail closed) rather than serve on an unverified balance', async () => {
+    const resolveTier = vi.fn(async () => 'pro');
+    const hasSpendableBalance = vi.fn(async () => true);
+    const decision = await resolveAppRoute(
+      'acme.pagespace.app',
+      deps({ resolvePayerId: async () => null, resolveTier, hasSpendableBalance }),
+    );
+    expect(decision).toEqual({ kind: 'parked', reason: 'out_of_credits' });
+    expect(resolveTier).not.toHaveBeenCalled();
+    expect(hasSpendableBalance).not.toHaveBeenCalled();
+  });
+
   it('given an insolvent payer, should park rather than replay', async () => {
     const decision = await resolveAppRoute(
       'acme.pagespace.app',
@@ -279,6 +303,14 @@ describe('defaultAppRouterDeps — the real edge is wired to the real readers',
     expect(tier).toBe('pro');
   });
 
+  // The identical function `app-billing.ts` hands the meter and the wake gate —
+  // not an equivalent bound the same way, the same reference — so "the router
+  // asks the wrong payer" is structurally impossible rather than a convention
+  // that can drift.
+  it('resolves the payer through the IDENTICAL function the meter and wake gate use', () => {
+    expect(defaultAppRouterDeps.resolvePayerId).toBe(defaultAppBillingDeps.resolvePayerId);
+  });
+
   // The row reader is module-private, so it is pinned by what it queries: the
   // published_apps table, keyed on `subdomain`, one row.
   it('reads the published_apps row for the subdomain it is asked about', async () => {

packages/lib/src/services/app-hosting/app-lifecycle-metering.ts

@@ -530,6 +530,11 @@ async function stopPublishedAppSerialized(
     await mirrorRecentFlyEvents(ref, deps);
   }
 
+  // NOT re-read here: the whole of `stopPublishedAppSerialized` runs under the
+  // awake meter's advisory lock (`stopPublishedApp` above), so no meter tick or
+  // concurrent stop can touch `row.awakeBilledThrough` between the read at the top
+  // of this function and the settle below — the lock is what makes the snapshot
+  // safe to settle against, not a fresh read racing to catch up with it.
   const settled = await settleAndClose(row, stoppedAt, nextStatus, stoppedAt, deps, reason);
   if (reason === 'daily_cap') reportDailyCapPark(row);
   return { outcome: 'stopped', status: nextStatus, billedSeconds: settled.billedSeconds };
@@ -703,6 +708,31 @@ async function settleAndClose(
   return { billedSeconds, failed: false };
 }
 
+/**
+ * Claim an abandoned tail exclusively: a CAS that clears `awakeBilledThrough`
+ * and `awakeHoldId` off the row, guarded on the exact watermark this caller read.
+ *
+ * A concurrent wake reads the same stale row and computes the same plan, so
+ * without this guard both would bill the same span. Only the first write to land
+ * matches the guard — `awakeBilledThrough` on the row still equals what was read
+ * — and clears it; the second finds the column already `null` and matches no
+ * row, learning it lost the race before it ever calls `trackUsage`.
+ */
+async function claimAbandonedTail(row: PublishedApp): Promise<boolean> {
+  const [claimed] = await db
+    .update(publishedApps)
+    .set({ awakeBilledThrough: null, awakeHoldId: null })
+    .where(
+      and(
+        eq(publishedApps.id, row.id),
+        eq(publishedApps.status, row.status),
+        eq(publishedApps.awakeBilledThrough, row.awakeBilledThrough as Date),
+      ),
+    )
+    .returning({ id: publishedApps.id });
+  return !!claimed;
+}
+
 /**
  * Settle a tail that a FAILED close left behind on a non-running row, at the wake
  * that is about to overwrite it.
@@ -731,6 +761,17 @@ async function settleAbandonedTail(
   if (row.awakeBilledThrough === null || row.lastStopAt === null) return;
 
   const plan = planAwakeSettle({ billedThrough: row.awakeBilledThrough, now: row.lastStopAt });
+
+  // Claim the tail EXCLUSIVELY, before doing anything with it: a guarded CAS that
+  // clears the watermark (and the hold riding with it) off the row. Whoever wins
+  // this write is the only caller that may bill or release this tail — a second
+  // wake racing this one, or this same wake retried after a `start_failed`, reads
+  // `awakeBilledThrough: null` and returns above before ever reaching here. That
+  // is what makes the tail settle at-most-once instead of "once per wake attempt
+  // until one finally succeeds".
+  const claimed = await claimAbandonedTail(row);
+  if (!claimed) return; // lost the race, or a previous attempt already cleared it
+
   if (plan.action !== 'settle') {
     // Nothing billable was stranded; return the reservation the failed close kept.
     if (row.awakeHoldId) await deps.billing.releaseHold(row.awakeHoldId);
@@ -739,8 +780,9 @@ async function settleAbandonedTail(
 
   const payerId = await deps.billing.resolvePayerId({ driveId: row.driveId });
   if (!payerId) {
-    // Unresolvable drive — never substitute a payer. The tail stays on the row and
-    // the wake below overwrites it, so this IS the loss; say so rather than warn.
+    // Unresolvable drive — never substitute a payer. The watermark is already
+    // claimed off the row above, so there is no later retry that could recover
+    // this span; say so at ERROR rather than warn.
     loggers.ai.error(
       'Published-app abandoned tail could not be billed: the owning drive did not resolve — this span is lost',
       new Error('abandoned tail payer unresolved'),