Persist only changed emulator state and roll up Autumn usage

packages/@emulators/autumn/src/routes/api.ts

@@ -11,6 +11,7 @@ import {
   applyCancelAction,
   balanceForFeature,
   checkAndConsume,
+  compactUsage,
   hasBillingCycle,
   knownFeature,
   type CancelAction,
@@ -98,12 +99,13 @@ export function autumnApiRoutes(ctx: RouteContext): void {
     if (!customerId || !featureId) {
       return c.json({ message: "customer_id and feature_id are required", code: "invalid_request" }, 400);
     }
-    ensureCustomer(as(), customerId, {});
+    const customer = ensureCustomer(as(), customerId, {});
     const event = as().events.insert({
       customer_id: customerId,
       feature_id: featureId,
       value: typeof body.value === "number" ? body.value : 1,
     });
+    compactUsage(as(), customer, featureId);
     return c.json({
       id: `evt_emulate_${event.id}`,
       code: "event_received",
@@ -207,6 +209,7 @@ export function autumnApiRoutes(ctx: RouteContext): void {
     const delta = targetUsage - balance.usage;
     if (delta !== 0) {
       store.events.insert({ customer_id: customerId, feature_id: featureId, value: delta });
+      compactUsage(store, customer, featureId);
     }
     return c.json({ success: true });
   });

packages/@emulators/autumn/src/serialize.ts

@@ -14,6 +14,7 @@ import type {
   AutumnPlan,
   AutumnPlanItem,
   AutumnSubscription,
+  AutumnTrackEvent,
 } from "./entities.js";
 
 const DAY_MS = 86_400_000;
@@ -177,19 +178,71 @@ function resetWindow(sub: AutumnSubscription, item: AutumnPlanItem): { start: nu
   return { start, resetsAt: start + span };
 }
 
+function eventTime(event: AutumnTrackEvent): number {
+  return Date.parse(event.created_at) || 0;
+}
+
 function usageFor(as: AutumnStore, customerId: string, featureId: string, since: number, after: number): number {
   return as.events
-    .all()
-    .filter(
-      (e) =>
-        e.customer_id === customerId &&
-        e.feature_id === featureId &&
-        e.id > after &&
-        (Date.parse(e.created_at) || 0) >= since,
-    )
+    .findBy("customer_id", customerId)
+    .filter((e) => e.feature_id === featureId && e.id > after && eventTime(e) >= since)
     .reduce((sum, e) => sum + (e.value ?? 0), 0);
 }
 
+/** Raw events one customer and feature may hold before they are rolled up. */
+const USAGE_ROLLUP_THRESHOLD = 64;
+
+/**
+ * Roll up a customer's usage events for one feature so that storage and balance
+ * reads stay bounded however many executions are tracked. Real Autumn keeps a
+ * running balance per window rather than replaying history, so a rollup is as
+ * faithful as the raw events for every balance Autumn can report.
+ *
+ * Balances count events after a subscription's usage watermark (by event id)
+ * and inside an item's current reset window (by time). Only adjacent events that
+ * no watermark or current window start separates are merged; the merged event
+ * keeps the newest id and time, so every current balance is unchanged. Later
+ * windows start after every existing event, and later watermarks are at or
+ * above every existing id, so those balances are unchanged too. A catalog edit
+ * that adds a reset interval to an existing item sees rolled-up history at the
+ * granularity of the rollups. `events.list` returns the rollups.
+ */
+export function compactUsage(as: AutumnStore, customer: AutumnCustomer, featureId: string): void {
+  const events = as.events.findBy("customer_id", customer.customer_id).filter((e) => e.feature_id === featureId);
+  if (events.length <= USAGE_ROLLUP_THRESHOLD) return;
+  const watermarks: number[] = [];
+  const windowStarts: number[] = [];
+  for (const sub of customer.subscriptions ?? []) {
+    watermarks.push(sub.usage_epoch ?? 0);
+    const plan = as.plans.findOneBy("plan_id", sub.plan_id);
+    for (const item of plan?.items ?? []) {
+      if (item.feature_id !== featureId) continue;
+      const window = resetWindow(sub, item);
+      if (window) windowStarts.push(window.start);
+    }
+  }
+  const segment = (event: AutumnTrackEvent) =>
+    `${watermarks.filter((mark) => event.id > mark).length}:${windowStarts.filter((start) => eventTime(event) >= start).length}`;
+  const merge = (run: AutumnTrackEvent[]) => {
+    if (run.length < 2) return;
+    const last = run[run.length - 1];
+    as.events.update(last.id, { value: run.reduce((sum, e) => sum + (e.value ?? 0), 0) });
+    for (const event of run.slice(0, -1)) as.events.delete(event.id);
+  };
+  let run: AutumnTrackEvent[] = [];
+  let runSegment = "";
+  for (const event of events.sort((a, b) => a.id - b.id)) {
+    const key = segment(event);
+    if (run.length > 0 && key !== runSegment) {
+      merge(run);
+      run = [];
+    }
+    run.push(event);
+    runSegment = key;
+  }
+  merge(run);
+}
+
 /** The emulator synthesizes this object for autumn-js's `customerToFeatures`
  *  helper, which throws unless every `balances` entry carries a nested
  *  `feature`. The real v1 API at api-version 2.3.0 omits it; it is an additive
@@ -369,7 +422,9 @@ export function checkAndConsume(
   const allowed = balance.unlimited || balance.overage_allowed || balance.remaining >= requiredBalance;
   if (!allowed || !sendEvent || requiredBalance === 0) return { allowed, balance };
   as.events.insert({ customer_id: customer.customer_id, feature_id: featureId, value: requiredBalance });
-  return { allowed, balance: balanceForFeature(as, customer, featureId) ?? balance };
+  const consumed = balanceForFeature(as, customer, featureId) ?? balance;
+  compactUsage(as, customer, featureId);
+  return { allowed, balance: consumed };
 }
 
 /** Mint the next Stripe-style PaymentMethod id for this instance. Stripe ids

packages/@emulators/cloudflare/src/durable-object.ts

@@ -100,12 +100,19 @@ const ledgerEntryKey = (id: string): string => `${LEDGER_ENTRY_PREFIX}${encodeKe
 
 // One Durable Object instance == one stateful emulator instance. Its `store`
 // lives in DO memory (single-threaded → the serialized-write consistency the
-// in-process emulator already assumes), snapshotted to DO storage after every
-// mutating request so the instance survives eviction. Auth is FAITHFUL by
+// in-process emulator already assumes). After every mutating request, the
+// changed parts of the store and ledger are written to DO storage so the
+// instance survives eviction. Auth is FAITHFUL by
 // default (strict): only seeded or minted tokens work; everything else gets the
 // real API's 401/403. Mint tokens at runtime via `POST /__token`.
 export class EmulatorDurableObject {
   private live?: Live;
+  // The JSON of every snapshot and ledger value last written to storage, by key.
+  // `persist()` writes only the keys whose value changed and deletes only the
+  // keys the state no longer has, so a request that changed nothing costs no
+  // storage writes and the cost of a request does not grow with stored history.
+  // Undefined until read from storage for the current live instance.
+  private written?: Map<string, string>;
 
   constructor(
     private readonly state: DurableObjectState,
@@ -295,6 +302,7 @@ export class EmulatorDurableObject {
     const entry = SERVICES[service];
     if (!entry) throw new Error(`unknown emulator service: ${service}`);
 
+    this.written = undefined;
     const persisted = await this.readPersistedState();
     await this.migrateLegacyState(persisted);
     const strict = persisted.strict ?? true;
@@ -374,15 +382,63 @@ export class EmulatorDurableObject {
     return this.live;
   }
 
+  /** Every snapshot and ledger value the live state needs in storage, by key. */
+  private persistedEntries(live: Live): Map<string, unknown> {
+    const entries = new Map<string, unknown>();
+    const snapshot = live.store.snapshot();
+    const meta: SnapshotMeta = { collections: {} };
+    for (const [key, value] of Object.entries(snapshot.data)) entries.set(snapshotDataKey(key), value);
+    for (const [name, collection] of Object.entries(snapshot.collections)) {
+      meta.collections[name] = { autoId: collection.autoId, indexFields: collection.indexFields };
+      for (const item of collection.items) entries.set(snapshotItemKey(name, item.id), item);
+    }
+    entries.set(SNAPSHOT_META_KEY, meta);
+    // The request ledger is intentionally durable, agent-readable history. Entries
+    // are split by id so each request writes only its own entry.
+    const ledger = live.ledger.serialize();
+    for (const entry of ledger.entries) entries.set(ledgerEntryKey(entry.id), entry);
+    entries.set(LEDGER_META_KEY, {
+      counter: ledger.counter,
+      ids: ledger.entries.map((entry) => entry.id),
+    } satisfies LedgerMeta);
+    return entries;
+  }
+
+  private async readWritten(): Promise<Map<string, string>> {
+    const written = new Map<string, string>();
+    for (const prefix of [SNAPSHOT_ITEM_PREFIX, SNAPSHOT_DATA_PREFIX, LEDGER_ENTRY_PREFIX]) {
+      for (const [key, value] of await this.state.storage.list({ prefix })) written.set(key, JSON.stringify(value));
+    }
+    for (const key of [SNAPSHOT_META_KEY, LEDGER_META_KEY]) {
+      const value = await this.state.storage.get(key);
+      if (value !== undefined) written.set(key, JSON.stringify(value));
+    }
+    return written;
+  }
+
   private async persist(): Promise<void> {
-    if (!this.live) return;
-    const persisted = await this.readPersistedState();
-    await this.writeStoreSnapshot(this.live.store.snapshot());
-    // Persist the request ledger so inspection history survives Durable Object
-    // eviction. Entries are split by id because the ledger is intentionally
-    // durable, agent-readable history.
-    await this.writeLedger(this.live.ledger.serialize());
-    await this.writeStateMeta(persisted);
+    const live = this.live;
+    if (!live) return;
+    const written = (this.written ??= await this.readWritten());
+    const puts: Array<{ key: string; value: unknown }> = [];
+    const entries = this.persistedEntries(live);
+    for (const [key, value] of entries) {
+      const json = JSON.stringify(value);
+      if (written.get(key) === json) continue;
+      puts.push({ key, value });
+      written.set(key, json);
+    }
+    const deletes = [...written.keys()].filter((key) => !entries.has(key));
+    for (const key of deletes) written.delete(key);
+    if (puts.length === 0 && deletes.length === 0) return;
+    try {
+      await inBatches(puts, (put) => this.state.storage.put(put.key, put.value));
+      await this.deleteKeys(deletes);
+    } catch (error) {
+      // Storage no longer matches the recorded values; read it again next time.
+      this.written = undefined;
+      throw error;
+    }
   }
 
   async fetch(request: Request): Promise<Response> {
@@ -408,6 +464,7 @@ export class EmulatorDurableObject {
       await this.clearMinted();
       await this.writeStateMeta({ seed, strict: strict !== false });
       this.live = undefined;
+      this.written = undefined;
       await this.ensure(service, instance, baseUrl);
       await this.persist();
       return Response.json({ ok: true, url: baseUrl, strict: strict !== false });

skills/autumn/SKILL.md

@@ -125,6 +125,8 @@ Attaching a plan after an immediate cancel issues fresh grants, so usage starts
 
 `balances.update` sets a customer's balance for one feature. Exactly one of `usage`, `remaining`, or `add_to_balance` is required. Use it for continuous-use features (seats, storage) where the app reconciles an absolute count rather than tracking deltas. The update is recorded as an adjustment event, so `events.list` shows the reconciliation and `balances.check` stays consistent. Unknown customers 404 with `customer_not_found`; a feature the customer's plan does not carry 404s with `not_found`.
 
+Once a customer has more than 64 usage events for a feature, the emulator rolls up adjacent events into one event carrying their total. It only rolls up events that no subscription watermark or current reset window separates, so every balance stays the same. `events.list` then returns the rolled-up events, and storage stays bounded however many checks consume usage.
+
 ```ts
 await autumn.balances.update({ customerId: "org_123", featureId: "members", usage: 12 });
 ```