Merge pull request 'Stop simultaneous transfers from overdrawing a branch' (#4) from fix/transfer-race into main

Reviewed-on: #4

docs/PILOT_READINESS.md

@@ -62,7 +62,7 @@ the decisions taken along the way, and what is still open.
 | Decision | Why |
 |---|---|
 | Offline sales are always accepted, even if they take stock below zero. | The sale already happened. Refusing it would lose real cash. Negative stock is shown on the Stock page and fixed with a physical count. |
-| Transfers are refused when the sending branch lacks the stock. | Transfers are made online by a manager, so they can be checked. |
+| Transfers are refused when the sending branch lacks the stock. | Transfers are made online by a manager, so they can be checked. The check and the move run under a short per-branch, per-product lock (`withStockLock` in `src/lib/server/stock.ts`), so simultaneous transfers cannot together overdraw a branch; physical counts take the same lock. |
 | The server price is authoritative; a different price shown on the device is kept (`clientUnitPrice`) and flagged (`priceMismatch`). | Trusting the device's price would let a cashier under-charge. Overwriting it silently would hide a cash discrepancy. |
 | A queued sale stays with the branch where it was recorded, even if the cashier has since moved. | That is where the stock left and the cash was taken. |
 | A product deactivated while a phone was offline is still recorded at its price. | Same reason as the first row: the sale happened. |
@@ -76,7 +76,7 @@ the decisions taken along the way, and what is still open.
   sales from the old version are sent on the next sync.
 
 ## Testing
-- `npm test` runs 56 API tests against an in-memory MongoDB, covering role
+- `npm test` runs 62 API tests against an in-memory MongoDB, covering role
   enforcement, tenant isolation for sales, stock and transfers, suspension,
   the sync protocol (including a 120-sale backlog replayed twice), and the
   business day.
@@ -137,8 +137,6 @@ prices read-only.
 - **Isolation by convention.** Tenant filters are still written per route.
   The tests cover the sales, stock, transfer and access paths, not every
   route.
-- **Concurrent transfers.** Two transfers submitted at the same moment
-  could together send more than the branch holds.
 - **Cost of goods is an average.** It uses each product's average buying
   price across the business, not first-in-first-out per branch. Units sold
   with no purchase on record are counted and flagged, but have no cost.

src/app/api/stock/adjustments/route.ts

@@ -6,7 +6,7 @@ import { Product } from "@/lib/server/models/Product";
 import { StockMovement } from "@/lib/server/models/StockMovement";
 import { authorize, MANAGERS } from "@/lib/server/auth";
 import { objectId } from "@/lib/server/validation";
-import { clientRefId, recordMovement, stockOf } from "@/lib/server/stock";
+import { clientRefId, recordMovement, stockOf, StockBusyError, withStockLock } from "@/lib/server/stock";
 
 export const runtime = "nodejs";
 
@@ -57,27 +57,36 @@ export async function POST(req: NextRequest) {
   if (!branch) return NextResponse.json({ error: "Branch not found in your business" }, { status: 404 });
   if (!product) return NextResponse.json({ error: "Product not found in your business" }, { status: 404 });
 
-  // A retried submit with the same clientUuid is a no-op.
+  // A retried submit with the same clientUuid is a no-op. A count reads the
+  // balance to work out its correction, so it runs under the stock lock:
+  // a transfer cannot slip in between the read and the write.
   const refId = clientRefId(auth.businessId, data.clientUuid);
-  const already = await StockMovement.exists({ type: "ADJUSTMENT", refId });
+  const lockedBranch = branchId;
-  if (!already) {
+  try {
-    const before = await stockOf(auth.businessId, branchId, data.productId);
+    const quantity = await withStockLock(auth.businessId, lockedBranch, data.productId, async () => {
-    const delta = data.mode === "COUNT" ? data.quantity - before : data.quantity;
+      const already = await StockMovement.exists({ type: "ADJUSTMENT", refId });
-    if (delta !== 0) {
+      if (!already) {
-      const label = data.mode === "COUNT" ? `Count: ${before} -> ${data.quantity}` : "Adjustment";
+        const before = await stockOf(auth.businessId, lockedBranch, data.productId);
-      await recordMovement({
+        const delta = data.mode === "COUNT" ? data.quantity - before : data.quantity;
-        businessId: auth.businessId,
+        if (delta !== 0) {
-        branchId,
+          const label = data.mode === "COUNT" ? `Count: ${before} -> ${data.quantity}` : "Adjustment";
-        productId: data.productId,
+          await recordMovement({
-        type: "ADJUSTMENT",
+            businessId: auth.businessId,
-        quantity: delta,
+            branchId: lockedBranch,
-        refId,
+            productId: data.productId,
-        note: data.note ? `${label}. ${data.note}` : label,
+            type: "ADJUSTMENT",
-        recordedBy: auth.sub,
+            quantity: delta,
-      });
+            refId,
-    }
+            note: data.note ? `${label}. ${data.note}` : label,
+            recordedBy: auth.sub,
+          });
+        }
+      }
+      return stockOf(auth.businessId, lockedBranch, data.productId);
+    });
+    return NextResponse.json({ branchId, productId: data.productId, quantity }, { status: 201 });
+  } catch (err) {
+    if (err instanceof StockBusyError) return NextResponse.json({ error: err.message }, { status: 409 });
+    throw err;
   }
-
-  const quantity = await stockOf(auth.businessId, branchId, data.productId);
-  return NextResponse.json({ branchId, productId: data.productId, quantity }, { status: 201 });
 }

src/app/api/stock/transfers/route.ts

@@ -6,7 +6,7 @@ import { Product } from "@/lib/server/models/Product";
 import { StockTransfer, IStockTransfer } from "@/lib/server/models/StockTransfer";
 import { authorize, branchParam, MANAGERS } from "@/lib/server/auth";
 import { objectId } from "@/lib/server/validation";
-import { recordMovement, stockOf } from "@/lib/server/stock";
+import { recordMovement, stockOf, StockBusyError, withStockLock } from "@/lib/server/stock";
 
 export const runtime = "nodejs";
 export const dynamic = "force-dynamic";
@@ -104,33 +104,49 @@ export async function POST(req: NextRequest) {
   if (!product) return NextResponse.json({ error: "Product not found in your business" }, { status: 404 });
 
   // Transfers are made online by a manager, so unlike offline sales they
-  // can be refused when the sending branch does not have the stock.
+  // can be refused when the sending branch does not have the stock. The
-  const available = await stockOf(auth.businessId, fromBranchId, data.productId);
+  // balance check and the movements run under the sending branch's stock
-  if (available < data.quantity) {
+  // lock, so two simultaneous transfers cannot both spend the same units.
-    return NextResponse.json(
+  const sendingBranch = fromBranchId;
-      { error: `Only ${Math.max(available, 0)} in stock at the sending branch`, available },
-      { status: 409 }
-    );
-  }
-
-  let transfer: IStockTransfer;
   try {
-    transfer = await StockTransfer.create({
+    return await withStockLock(auth.businessId, sendingBranch, data.productId, async () => {
-      businessId: auth.businessId,
+      // An identical submit may have completed while we waited for the lock.
-      fromBranchId,
+      const done = await StockTransfer.findOne({ businessId: auth.businessId, clientUuid: data.clientUuid });
-      toBranchId: data.toBranchId,
+      if (done) {
-      productId: data.productId,
+        await applyTransfer(done);
-      quantity: data.quantity,
+        return NextResponse.json({ transfer: done }, { status: 200 });
-      clientUuid: data.clientUuid,
+      }
-      note: data.note ?? null,
-      recordedBy: auth.sub,
-    });
-  } catch (err: unknown) {
-    // Two identical submits raced and the other one created it.
-    if ((err as { code?: number })?.code !== 11000) throw err;
-    transfer = (await StockTransfer.findOne({ businessId: auth.businessId, clientUuid: data.clientUuid }))!;
-  }
-  await applyTransfer(transfer);
 
-  return NextResponse.json({ transfer }, { status: 201 });
+      const available = await stockOf(auth.businessId, sendingBranch, data.productId);
+      if (available < data.quantity) {
+        return NextResponse.json(
+          { error: `Only ${Math.max(available, 0)} in stock at the sending branch`, available },
+          { status: 409 }
+        );
+      }
+
+      let transfer: IStockTransfer;
+      try {
+        transfer = await StockTransfer.create({
+          businessId: auth.businessId,
+          fromBranchId: sendingBranch,
+          toBranchId: data.toBranchId,
+          productId: data.productId,
+          quantity: data.quantity,
+          clientUuid: data.clientUuid,
+          note: data.note ?? null,
+          recordedBy: auth.sub,
+        });
+      } catch (err: unknown) {
+        // Defensive: the unique clientUuid index still guards double submits.
+        if ((err as { code?: number })?.code !== 11000) throw err;
+        transfer = (await StockTransfer.findOne({ businessId: auth.businessId, clientUuid: data.clientUuid }))!;
+      }
+      await applyTransfer(transfer);
+      return NextResponse.json({ transfer }, { status: 201 });
+    });
+  } catch (err) {
+    if (err instanceof StockBusyError) return NextResponse.json({ error: err.message }, { status: 409 });
+    throw err;
+  }
 }

src/lib/server/models/StockLock.ts

@@ -0,0 +1,25 @@
+import { Schema, model, models, Model } from "mongoose";
+
+export interface IStockLock {
+  _id: string; // `${businessId}:${branchId}:${productId}`
+  owner: string; // random id of the request holding the lock
+  lockedUntil: Date; // lease expiry: a crashed request cannot hold the lock forever
+}
+
+/**
+ * A short lease on one product's stock at one branch, held while a request
+ * reads the current balance and writes a movement that depends on it (see
+ * withStockLock in src/lib/server/stock.ts). Rows are deleted on release;
+ * an expired row is simply taken over by the next request.
+ */
+const stockLockSchema = new Schema<IStockLock>(
+  {
+    _id: { type: String, required: true },
+    owner: { type: String, required: true },
+    lockedUntil: { type: Date, required: true },
+  },
+  { versionKey: false }
+);
+
+export const StockLock =
+  (models.StockLock as Model<IStockLock>) || model<IStockLock>("StockLock", stockLockSchema);

src/lib/server/stock.ts

@@ -3,6 +3,7 @@ import { StockMovement, type MovementType } from "@/lib/server/models/StockMovem
 import { ACTIVE_BRANCH, Branch } from "@/lib/server/models/Branch";
 import { Product } from "@/lib/server/models/Product";
 import { Purchase } from "@/lib/server/models/Purchase";
+import { StockLock } from "@/lib/server/models/StockLock";
 import type { ITransaction } from "@/lib/server/models/Transaction";
 
 type Id = Types.ObjectId | string;
@@ -173,3 +174,58 @@ export async function backfillPurchaseMovements(businessId?: Id): Promise<number
   }
   return added;
 }
+
+const LOCK_LEASE_MS = 10_000;
+const LOCK_WAIT_MS = 5_000;
+
+/** Another request kept the stock busy for longer than we are willing to wait. */
+export class StockBusyError extends Error {
+  constructor() {
+    super("This stock is being updated by someone else. Please try again.");
+  }
+}
+
+/**
+ * Run `fn` while holding the lock on one product's stock at one branch.
+ *
+ * Anything that reads the balance and then writes a movement based on it
+ * (a transfer refusing to overdraw, a physical count computing its
+ * correction) must run inside this. Without it, two transfers that each
+ * see 10 in stock can both send 6, because the read and the write are
+ * separate operations. With it, the second waits for the first and then
+ * sees 4.
+ *
+ * The lock is a lease row: acquired with one atomic upsert that only
+ * succeeds if no unexpired lease exists (otherwise the insert hits the
+ * unique _id and fails), released by deleting it. Sales do not take the
+ * lock - they are recorded even when stock runs out (see PILOT_READINESS).
+ */
+export async function withStockLock<T>(businessId: Id, branchId: Id, productId: Id, fn: () => Promise<T>): Promise<T> {
+  const key = `${businessId}:${branchId}:${productId}`;
+  const owner = crypto.randomUUID();
+  const deadline = Date.now() + LOCK_WAIT_MS;
+  let delay = 10;
+
+  for (;;) {
+    const now = Date.now();
+    try {
+      await StockLock.updateOne(
+        { _id: key, lockedUntil: { $lt: new Date(now) } },
+        { $set: { owner, lockedUntil: new Date(now + LOCK_LEASE_MS) } },
+        { upsert: true }
+      );
+      break;
+    } catch (err: unknown) {
+      if ((err as { code?: number })?.code !== 11000) throw err;
+    }
+    if (Date.now() > deadline) throw new StockBusyError();
+    await new Promise((r) => setTimeout(r, delay + Math.random() * delay));
+    delay = Math.min(delay * 2, 200);
+  }
+
+  try {
+    return await fn();
+  } finally {
+    await StockLock.deleteOne({ _id: key, owner });
+  }
+}

tests/concurrency.test.ts

@@ -0,0 +1,119 @@
+import { describe, expect, it } from "vitest";
+import { stockOf } from "@/lib/server/stock";
+import { StockTransfer } from "@/lib/server/models/StockTransfer";
+import { StockLock } from "@/lib/server/models/StockLock";
+import * as adjustments from "@/app/api/stock/adjustments/route";
+import * as transfers from "@/app/api/stock/transfers/route";
+import { makeTenant, req, uuid, type Tenant } from "./helpers";
+
+async function openingStock(t: Tenant, quantity: number) {
+  await adjustments.POST(
+    req("POST", "/x", {
+      as: t.owner,
+      body: { clientUuid: uuid(), branchId: t.branchA, productId: t.water20, mode: "DELTA", quantity },
+    })
+  );
+}
+
+const transferOut = (t: Tenant, quantity: number) =>
+  transfers.POST(
+    req("POST", "/x", {
+      as: t.owner,
+      body: { clientUuid: uuid(), fromBranchId: t.branchA, toBranchId: t.branchB, productId: t.water20, quantity },
+    })
+  );
+
+describe("concurrent stock changes", () => {
+  it("never lets simultaneous transfers send more than the branch holds", async () => {
+    const t = await makeTenant();
+    await openingStock(t, 10);
+
+    // Five managers press "Record transfer" for 4 units at the same moment.
+    const statuses = (await Promise.all(Array.from({ length: 5 }, () => transferOut(t, 4)))).map((r) => r.status);
+
+    expect(statuses.filter((s) => s === 201)).toHaveLength(2);
+    expect(statuses.filter((s) => s === 409)).toHaveLength(3);
+    expect(await stockOf(t.businessId, t.branchA, t.water20)).toBe(2);
+    expect(await stockOf(t.businessId, t.branchB, t.water20)).toBe(8);
+    expect(await StockTransfer.countDocuments({})).toBe(2);
+  });
+
+  it("holds across repeated races, not just once", async () => {
+    for (let round = 0; round < 5; round++) {
+      const t = await makeTenant("Round" + round);
+      await openingStock(t, 9);
+      const statuses = (await Promise.all(Array.from({ length: 4 }, () => transferOut(t, 3)))).map((r) => r.status);
+      expect(statuses.filter((s) => s === 201)).toHaveLength(3);
+      expect(await stockOf(t.businessId, t.branchA, t.water20)).toBe(0);
+    }
+  });
+
+  it("applies a double-tapped transfer once even when both taps race", async () => {
+    const t = await makeTenant();
+    await openingStock(t, 10);
+    const body = { clientUuid: uuid(), fromBranchId: t.branchA, toBranchId: t.branchB, productId: t.water20, quantity: 6 };
+    const statuses = (
+      await Promise.all([
+        transfers.POST(req("POST", "/x", { as: t.owner, body })),
+        transfers.POST(req("POST", "/x", { as: t.owner, body })),
+      ])
+    ).map((r) => r.status);
+    expect(statuses.sort()).toEqual([200, 201]);
+    expect(await stockOf(t.businessId, t.branchA, t.water20)).toBe(4);
+  });
+
+  it("takes over a lock left behind by a crashed request once it expires", async () => {
+    const t = await makeTenant();
+    await openingStock(t, 10);
+    await StockLock.create({
+      _id: `${t.businessId}:${t.branchA}:${t.water20}`,
+      owner: "crashed-request",
+      lockedUntil: new Date(Date.now() - 1000),
+    });
+    expect((await transferOut(t, 3)).status).toBe(201);
+    expect(await StockLock.countDocuments({})).toBe(0);
+  });
+
+  it("serialises a physical count with a transfer, so the count is never applied to a stale balance", async () => {
+    const t = await makeTenant();
+    await openingStock(t, 10);
+
+    // A count of 10 and a transfer of 6 race. Whichever runs second must
+    // see the first: either the count resets 4 back to 10 after the
+    // transfer, or the transfer moves 6 out of the counted 10.
+    const [count, transfer] = await Promise.all([
+      adjustments.POST(
+        req("POST", "/x", {
+          as: t.owner,
+          body: { clientUuid: uuid(), branchId: t.branchA, productId: t.water20, mode: "COUNT", quantity: 10 },
+        })
+      ),
+      transferOut(t, 6),
+    ]);
+    expect([count.status, transfer.status]).toEqual([201, 201]);
+    const a = await stockOf(t.businessId, t.branchA, t.water20);
+    expect([4, 10]).toContain(a);
+    expect(await stockOf(t.businessId, t.branchB, t.water20)).toBe(6);
+  });
+
+  it("does not block transfers of different products or branches on each other", async () => {
+    const t = await makeTenant();
+    await openingStock(t, 10);
+    await adjustments.POST(
+      req("POST", "/x", {
+        as: t.owner,
+        body: { clientUuid: uuid(), branchId: t.branchA, productId: t.water5, mode: "DELTA", quantity: 10 },
+      })
+    );
+    const results = await Promise.all([
+      transferOut(t, 5),
+      transfers.POST(
+        req("POST", "/x", {
+          as: t.owner,
+          body: { clientUuid: uuid(), fromBranchId: t.branchA, toBranchId: t.branchB, productId: t.water5, quantity: 5 },
+        })
+      ),
+    ]);
+    expect(results.map((r) => r.status)).toEqual([201, 201]);
+  });
+});