feat(route): optimistic concurrency control for pair-meta updates

openapi.yaml

@@ -81,6 +81,19 @@ components:
           maximum: 1000
           description: Fee in basis points (0–1000).
           example: 30
+        enabled:
+          type: boolean
+          description: Whether this pair is enabled for quoting.
+          example: true
+        rate:
+          type: string
+          description: Base exchange rate for the pair.
+          example: "1.0"
+        version:
+          type: integer
+          minimum: 0
+          description: Optimistic-concurrency version. Returned on every read; must be supplied as expected_version on PATCH to prevent stale writes.
+          example: 0
 
     ApiKeyRecord:
       type: object
@@ -664,6 +677,10 @@ paths:
                   type: integer
                   minimum: 0
                   maximum: 1000
+                expected_version:
+                  type: integer
+                  minimum: 0
+                  description: Current version from GET /info. Omit to skip OCC check (not recommended).
       responses:
         "200":
           description: Updated pair metadata.
@@ -705,6 +722,10 @@ paths:
               properties:
                 minAmount:
                   type: string
+                expected_version:
+                  type: integer
+                  minimum: 0
+                  description: Current version from GET /info. Omit to skip OCC check (not recommended).
                   pattern: "^[0-9]{1,39}$"
       responses:
         "200":
@@ -747,6 +768,10 @@ paths:
               properties:
                 maxAmount:
                   type: string
+                expected_version:
+                  type: integer
+                  minimum: 0
+                  description: Current version from GET /info. Omit to skip OCC check (not recommended).
                   pattern: "^[1-9][0-9]{0,38}$"
       responses:
         "200":
@@ -789,6 +814,10 @@ paths:
               properties:
                 liquidity:
                   type: string
+                expected_version:
+                  type: integer
+                  minimum: 0
+                  description: Current version from GET /info. Omit to skip OCC check (not recommended).
                   pattern: "^[0-9]{1,39}$"
       responses:
         "200":

src/__tests__/index.test.ts

@@ -1010,6 +1010,7 @@ describe("StableRoute Backend", () => {
         maxAmount: "0",
         liquidity: "0",
         enabled: true,
+        version: 0,
       });
     });
 
@@ -1055,6 +1056,115 @@ describe("StableRoute Backend", () => {
     });
   });
 
+  describe("optimistic concurrency control (OCC) for pair-meta patches", () => {
+    it("returns version 0 in GET /info for a freshly registered pair", async () => {
+      await request(app)
+        .post("/api/v1/pairs")
+        .send({ source: "OCC1", destination: "META" });
+
+      const info = await request(app).get("/api/v1/pairs/OCC1/META/info");
+      expect(info.status).toBe(200);
+      expect(info.body.version).toBe(0);
+    });
+
+    it("increments version on every successful PATCH", async () => {
+      await request(app)
+        .post("/api/v1/pairs")
+        .send({ source: "OCC2", destination: "META" });
+
+      const patch1 = await request(app)
+        .patch("/api/v1/pairs/OCC2/META/fee_bps")
+        .send({ feeBps: 10 });
+      expect(patch1.status).toBe(200);
+      expect(patch1.body.version).toBe(1);
+
+      const patch2 = await request(app)
+        .patch("/api/v1/pairs/OCC2/META/fee_bps")
+        .send({ feeBps: 20 });
+      expect(patch2.status).toBe(200);
+      expect(patch2.body.version).toBe(2);
+
+      const info = await request(app).get("/api/v1/pairs/OCC2/META/info");
+      expect(info.body.version).toBe(2);
+    });
+
+    it("rejects a stale write with 409 when expected_version does not match", async () => {
+      await request(app)
+        .post("/api/v1/pairs")
+        .send({ source: "OCC3", destination: "META" });
+
+      // Bump version to 1
+      await request(app)
+        .patch("/api/v1/pairs/OCC3/META/fee_bps")
+        .send({ feeBps: 10 });
+
+      // Attempt stale write with expected_version 0
+      const stale = await request(app)
+        .patch("/api/v1/pairs/OCC3/META/fee_bps")
+        .send({ feeBps: 20, expected_version: 0 });
+      expect(stale.status).toBe(409);
+      expect(stale.body.error).toBe("conflict");
+      expect(stale.body.message).toMatch(/stale version/);
+    });
+
+    it("allows a concurrent-safe write when expected_version matches", async () => {
+      await request(app)
+        .post("/api/v1/pairs")
+        .send({ source: "OCC4", destination: "META" });
+
+      const info = await request(app).get("/api/v1/pairs/OCC4/META/info");
+      const v0 = info.body.version;
+
+      const patch = await request(app)
+        .patch("/api/v1/pairs/OCC4/META/fee_bps")
+        .send({ feeBps: 30, expected_version: v0 });
+      expect(patch.status).toBe(200);
+      expect(patch.body.version).toBe(v0 + 1);
+    });
+
+    it("rejects non-numeric expected_version with 400", async () => {
+      await request(app)
+        .post("/api/v1/pairs")
+        .send({ source: "OCC5", destination: "META" });
+
+      const res = await request(app)
+        .patch("/api/v1/pairs/OCC5/META/fee_bps")
+        .send({ feeBps: 10, expected_version: "zero" as unknown });
+      expect(res.status).toBe(400);
+      expect(res.body.error).toBe("invalid_request");
+    });
+
+    it("rejects unknown keys alongside expected_version", async () => {
+      await request(app)
+        .post("/api/v1/pairs")
+        .send({ source: "OCC6", destination: "META" });
+
+      const res = await request(app)
+        .patch("/api/v1/pairs/OCC6/META/fee_bps")
+        .send({ feeBps: 10, expected_version: 0, extra: true });
+      expect(res.status).toBe(400);
+      expect(res.body.unknownKeys).toContain("extra");
+    });
+
+    it("resets version to 0 on POST /reset", async () => {
+      await request(app)
+        .post("/api/v1/pairs")
+        .send({ source: "OCC7", destination: "META" });
+
+      await request(app)
+        .patch("/api/v1/pairs/OCC7/META/fee_bps")
+        .send({ feeBps: 10 });
+
+      const reset = await request(app)
+        .post("/api/v1/pairs/OCC7/META/reset");
+      expect(reset.status).toBe(200);
+      expect(reset.body.version).toBe(0);
+
+      const info = await request(app).get("/api/v1/pairs/OCC7/META/info");
+      expect(info.body.version).toBe(0);
+    });
+  });
+
   describe("GET /api/v1/quote — pair registration requirement", () => {
     it("returns 404 pair_not_registered for an unregistered pair", async () => {
       const res = await request(app)
@@ -2128,6 +2238,115 @@ describe("StableRoute Backend", () => {
     });
   });
 
+  describe("optimistic concurrency control (OCC) for pair-meta patches", () => {
+    it("returns version 0 in GET /info for a freshly registered pair", async () => {
+      await request(app)
+        .post("/api/v1/pairs")
+        .send({ source: "OCC1", destination: "META" });
+
+      const info = await request(app).get("/api/v1/pairs/OCC1/META/info");
+      expect(info.status).toBe(200);
+      expect(info.body.version).toBe(0);
+    });
+
+    it("increments version on every successful PATCH", async () => {
+      await request(app)
+        .post("/api/v1/pairs")
+        .send({ source: "OCC2", destination: "META" });
+
+      const patch1 = await request(app)
+        .patch("/api/v1/pairs/OCC2/META/fee_bps")
+        .send({ feeBps: 10 });
+      expect(patch1.status).toBe(200);
+      expect(patch1.body.version).toBe(1);
+
+      const patch2 = await request(app)
+        .patch("/api/v1/pairs/OCC2/META/fee_bps")
+        .send({ feeBps: 20 });
+      expect(patch2.status).toBe(200);
+      expect(patch2.body.version).toBe(2);
+
+      const info = await request(app).get("/api/v1/pairs/OCC2/META/info");
+      expect(info.body.version).toBe(2);
+    });
+
+    it("rejects a stale write with 409 when expected_version does not match", async () => {
+      await request(app)
+        .post("/api/v1/pairs")
+        .send({ source: "OCC3", destination: "META" });
+
+      // Bump version to 1
+      await request(app)
+        .patch("/api/v1/pairs/OCC3/META/fee_bps")
+        .send({ feeBps: 10 });
+
+      // Attempt stale write with expected_version 0
+      const stale = await request(app)
+        .patch("/api/v1/pairs/OCC3/META/fee_bps")
+        .send({ feeBps: 20, expected_version: 0 });
+      expect(stale.status).toBe(409);
+      expect(stale.body.error).toBe("conflict");
+      expect(stale.body.message).toMatch(/stale version/);
+    });
+
+    it("allows a concurrent-safe write when expected_version matches", async () => {
+      await request(app)
+        .post("/api/v1/pairs")
+        .send({ source: "OCC4", destination: "META" });
+
+      const info = await request(app).get("/api/v1/pairs/OCC4/META/info");
+      const v0 = info.body.version;
+
+      const patch = await request(app)
+        .patch("/api/v1/pairs/OCC4/META/fee_bps")
+        .send({ feeBps: 30, expected_version: v0 });
+      expect(patch.status).toBe(200);
+      expect(patch.body.version).toBe(v0 + 1);
+    });
+
+    it("rejects non-numeric expected_version with 400", async () => {
+      await request(app)
+        .post("/api/v1/pairs")
+        .send({ source: "OCC5", destination: "META" });
+
+      const res = await request(app)
+        .patch("/api/v1/pairs/OCC5/META/fee_bps")
+        .send({ feeBps: 10, expected_version: "zero" as unknown });
+      expect(res.status).toBe(400);
+      expect(res.body.error).toBe("invalid_request");
+    });
+
+    it("rejects unknown keys alongside expected_version", async () => {
+      await request(app)
+        .post("/api/v1/pairs")
+        .send({ source: "OCC6", destination: "META" });
+
+      const res = await request(app)
+        .patch("/api/v1/pairs/OCC6/META/fee_bps")
+        .send({ feeBps: 10, expected_version: 0, extra: true });
+      expect(res.status).toBe(400);
+      expect(res.body.unknownKeys).toContain("extra");
+    });
+
+    it("resets version to 0 on POST /reset", async () => {
+      await request(app)
+        .post("/api/v1/pairs")
+        .send({ source: "OCC7", destination: "META" });
+
+      await request(app)
+        .patch("/api/v1/pairs/OCC7/META/fee_bps")
+        .send({ feeBps: 10 });
+
+      const reset = await request(app)
+        .post("/api/v1/pairs/OCC7/META/reset");
+      expect(reset.status).toBe(200);
+      expect(reset.body.version).toBe(0);
+
+      const info = await request(app).get("/api/v1/pairs/OCC7/META/info");
+      expect(info.body.version).toBe(0);
+    });
+  });
+
   describe("GET /api/v1/quote — pair registration requirement", () => {
     it("returns 404 pair_not_registered for an unregistered pair", async () => {
       const res = await request(app)

src/index.ts

@@ -2194,13 +2194,26 @@ const makePairMetaPatch =
       sendError(res, req, 404, "not_found", "pair not registered");
       return;
     }
-    if (rejectUnknownKeys(req, res, [bodyKey])) return;
-    const value = (req.body ?? {})[bodyKey] as unknown;
+    const body = req.body ?? {};
+    if (rejectUnknownKeys(req, res, [bodyKey, "expected_version"])) return;
+    const value = body[bodyKey] as unknown;
     if (!validate(value)) {
       sendError(res, req, 400, "invalid_request", errorMessage);
       return;
     }
     const meta = pairMeta.get(k) ?? defaultMeta();
+    // Optimistic-concurrency guard: reject stale writes
+    const expectedVersion = (body as Record<string, unknown>)["expected_version"];
+    if (expectedVersion !== undefined) {
+      if (typeof expectedVersion !== "number") {
+        sendError(res, req, 400, "invalid_request", "expected_version must be a number");
+        return;
+      }
+      if (expectedVersion !== meta.version) {
+        sendError(res, req, 409, "conflict", `stale version: expected ${expectedVersion}, found ${meta.version}`);
+        return;
+      }
+    }
     // Optional cross-field invariant (e.g. min <= max). Runs after the
     // per-field format check so `value` is already known to be a valid
     // integer string; comparisons stay in BigInt space (see crossCheck impls).
@@ -2212,6 +2225,7 @@ const makePairMetaPatch =
       }
     }
     (meta as Record<string, unknown>)[field] = value;
+    meta.version += 1;
     pairMeta.set(k, meta);
     invalidateQuoteCache(k);
     res.json({ source, destination, ...meta });

src/persistence.ts

@@ -20,7 +20,7 @@ import {
  * Increment when fields are added to {@link StoreSnapshot} so that older
  * snapshots can be upgraded via the migration chain in {@link migrateSnapshot}.
  */
-export const CURRENT_SCHEMA_VERSION = 1;
+export const CURRENT_SCHEMA_VERSION = 2;
 
 /**
  * Data structure representing a full snapshot of the in-memory stores.
@@ -127,6 +127,10 @@ export function migrateSnapshot(data: unknown): StoreSnapshot | null {
     snap = migrateV0ToV1(snap);
   }
 
+  if (version < 2) {
+    snap = migrateV1ToV2(snap);
+  }
+
   if (isValidSnapshot(snap)) {
     return snap as StoreSnapshot;
   }
@@ -167,6 +171,36 @@ function migrateV0ToV1(data: Record<string, unknown>): Record<string, unknown> {
   return data;
 }
 
+/**
+ * Migrate a version-1 snapshot to version 2.
+ *
+ * In version 1 the {@link PairMeta} type did not include the `version`
+ * field; this migration backfills `version: 0` so that older snapshots
+ * hydrate correctly.
+ */
+function migrateV1ToV2(data: Record<string, unknown>): Record<string, unknown> {
+  data.schemaVersion = 2;
+
+  if (Array.isArray(data.pairMeta)) {
+    for (let i = 0; i < data.pairMeta.length; i++) {
+      const entry = data.pairMeta[i];
+      if (
+        Array.isArray(entry) &&
+        entry.length === 2 &&
+        entry[1] !== null &&
+        typeof entry[1] === "object"
+      ) {
+        const meta = entry[1] as Record<string, unknown>;
+        if (typeof meta.version !== "number") {
+          meta.version = 0;
+        }
+      }
+    }
+  }
+
+  return data;
+}
+
 /**
  * Interface for pluggable persistence store adapters.
  */