fix(anpr): share one camera snapshot across bridge + advisory paths
On a vehicle entry, two paths captured the SAME Hikvision camera within ~1s — the ANPR bridge (barrier-driving) and the advisory snapshotAsync (evidence/ telemetry) — each from a separate adapter instance. Hikvision serves snapshots single-threaded, so the second concurrent GET returned HTTP 503; the bridge then fail-softed and burned its 12s debounce, producing a ~74s "slow" subscriber entry (observed 2026-06-25, Qazim Mulleti / AB816NN — plate read was instant at conf 1.000; the delay was the 503/debounce churn, not recognition). Add captureSnapshotShared() in snapshot.ts: a module-level, deviceId-keyed cache that both paths call. It coalesces in-flight captures (the 2nd caller awaits the 1st's pull → no concurrent 503), serves a brief freshness window (1500ms) so the bridge→advisory sequence for one vehicle reuses one frame, never caches a failure (next caller retries), and keys by deviceId (no cross-camera/stale-vehicle reuse). Wired into anpr-entry.ts (bridge) and snapshot.ts (advisory). Tests: snapshot.test.ts (concurrent coalescing, TTL reuse, TTL-lapse re-pull, failure-not-cached, per-camera keying); anpr-entry.test.ts mock updated. 168 server tests green. NOTE: this removes the latency (the 503 collision). The separate double-entry (two signed vehicle_entry for one car) — debounce-too-short / stamp-before- success — is still open; less likely now but not eliminated. Claude-Session: https://claude.ai/code/session_01Xcm6ikLgGoCxxHrxtjkk5V
This commit is contained in:
@@ -15,8 +15,13 @@ import type { SubscriptionFlow, SubscriptionMatch } from "./subscription-flow.js
|
|||||||
// Mock buildCamera so the bridge gets a fake camera whose captureSnapshot is a stub
|
// Mock buildCamera so the bridge gets a fake camera whose captureSnapshot is a stub
|
||||||
// (no registry, no network). The factory returns a fresh shot each call.
|
// (no registry, no network). The factory returns a fresh shot each call.
|
||||||
const captureSnapshot = vi.fn(async () => ({ bytes: Buffer.from("jpg"), contentType: "image/jpeg" }));
|
const captureSnapshot = vi.fn(async () => ({ bytes: Buffer.from("jpg"), contentType: "image/jpeg" }));
|
||||||
|
// The bridge now goes through captureSnapshotShared (the dedup wrapper, exercised in
|
||||||
|
// snapshot.test.ts); here it just delegates to the fake camera's captureSnapshot so this
|
||||||
|
// suite stays focused on the bridge's own match/debounce/emit logic.
|
||||||
vi.mock("./snapshot.js", () => ({
|
vi.mock("./snapshot.js", () => ({
|
||||||
buildCamera: () => ({ captureSnapshot }),
|
buildCamera: () => ({ captureSnapshot }),
|
||||||
|
captureSnapshotShared: (_id: string, camera: { captureSnapshot: typeof captureSnapshot }, ctx: unknown) =>
|
||||||
|
camera.captureSnapshot(ctx as never),
|
||||||
}));
|
}));
|
||||||
|
|
||||||
// Import AFTER the mock is registered.
|
// Import AFTER the mock is registered.
|
||||||
|
|||||||
@@ -3,7 +3,7 @@ import { devices, deviceEvents as deviceEventsTable, eq, siteConfig, type Db, ty
|
|||||||
import type { FastifyBaseLogger } from "fastify";
|
import type { FastifyBaseLogger } from "fastify";
|
||||||
import { deviceEvents, type DeviceReadEvent } from "./device-events.js";
|
import { deviceEvents, type DeviceReadEvent } from "./device-events.js";
|
||||||
import { directionOf, type FlowDirection } from "./device-resolve.js";
|
import { directionOf, type FlowDirection } from "./device-resolve.js";
|
||||||
import { buildCamera } from "./snapshot.js";
|
import { buildCamera, captureSnapshotShared } from "./snapshot.js";
|
||||||
import type { SubscriptionFlow } from "./subscription-flow.js";
|
import type { SubscriptionFlow } from "./subscription-flow.js";
|
||||||
import type { VisionClient } from "./vision-client.js";
|
import type { VisionClient } from "./vision-client.js";
|
||||||
|
|
||||||
@@ -100,7 +100,9 @@ export class AnprBridge {
|
|||||||
// "both" collapses to entry purely for the capture hint (it doesn't pick the lane —
|
// "both" collapses to entry purely for the capture hint (it doesn't pick the lane —
|
||||||
// the gated flow infers the verb from the camera's bound relay direction).
|
// the gated flow infers the verb from the camera's bound relay direction).
|
||||||
const direction: FlowDirection = directionOf(this.#db, row) === "exit" ? "exit" : "entry";
|
const direction: FlowDirection = directionOf(this.#db, row) === "exit" ? "exit" : "entry";
|
||||||
const shot = await camera.captureSnapshot({ direction });
|
// Shared capture (deviceId-keyed): coalesces with the advisory snapshotAsync for
|
||||||
|
// the SAME vehicle so the single-threaded camera isn't hit twice (→ HTTP 503).
|
||||||
|
const shot = await captureSnapshotShared(deviceId, camera, { direction });
|
||||||
const result = await this.#vision.analyze(shot.bytes, shot.contentType);
|
const result = await this.#vision.analyze(shot.bytes, shot.contentType);
|
||||||
if (!result || !result.plate) return; // nothing read
|
if (!result || !result.plate) return; // nothing read
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,93 @@
|
|||||||
|
import { describe, expect, it, vi } from "vitest";
|
||||||
|
import type { CameraDevice, Snapshot } from "@parking/devices";
|
||||||
|
import { captureSnapshotShared } from "./snapshot.js";
|
||||||
|
|
||||||
|
// captureSnapshotShared: one HTTP pull per camera per vehicle. A Hikvision unit serves
|
||||||
|
// snapshots SINGLE-THREADED (a 2nd concurrent GET → HTTP 503). On an entry the ANPR
|
||||||
|
// bridge AND the advisory snapshotAsync both capture the same camera within ~1s, each
|
||||||
|
// from a SEPARATE adapter instance — so this deviceId-keyed cache coalesces in-flight
|
||||||
|
// captures and serves a brief freshness window, collapsing the two into one real pull.
|
||||||
|
// (Root cause of the slow 2026-06-25 subscriber entry.)
|
||||||
|
|
||||||
|
/** A fake camera whose captureSnapshot is controllable (count calls, delay, fail). */
|
||||||
|
function fakeCamera(opts: { delayMs?: number; fail?: boolean; tag?: string } = {}): {
|
||||||
|
camera: CameraDevice;
|
||||||
|
calls: () => number;
|
||||||
|
} {
|
||||||
|
let calls = 0;
|
||||||
|
const tag = opts.tag ?? "x";
|
||||||
|
const camera = {
|
||||||
|
async captureSnapshot(): Promise<Snapshot> {
|
||||||
|
calls++;
|
||||||
|
if (opts.delayMs) await new Promise((r) => setTimeout(r, opts.delayMs));
|
||||||
|
if (opts.fail) throw new Error("HTTP 503");
|
||||||
|
// Tag distinguishes frames from different cameras (the per-camera keying test).
|
||||||
|
return { bytes: Buffer.from(`shot-${tag}-${calls}`), contentType: "image/jpeg", capturedAt: new Date().toISOString() };
|
||||||
|
},
|
||||||
|
} as unknown as CameraDevice;
|
||||||
|
return { camera, calls: () => calls };
|
||||||
|
}
|
||||||
|
|
||||||
|
/** A unique deviceId per test so the module-level cache never bleeds across cases. */
|
||||||
|
function id(): string {
|
||||||
|
return `cam-${Math.random().toString(36).slice(2)}`;
|
||||||
|
}
|
||||||
|
|
||||||
|
describe("captureSnapshotShared", () => {
|
||||||
|
it("coalesces CONCURRENT captures into a single hardware pull (the 503 fix)", async () => {
|
||||||
|
const { camera, calls } = fakeCamera({ delayMs: 20 });
|
||||||
|
const dev = id();
|
||||||
|
// The bridge and the advisory path fire at nearly the same instant.
|
||||||
|
const [a, b] = await Promise.all([
|
||||||
|
captureSnapshotShared(dev, camera, { direction: "entry" }),
|
||||||
|
captureSnapshotShared(dev, camera, { direction: "entry" }),
|
||||||
|
]);
|
||||||
|
expect(calls()).toBe(1); // ONE GET, not two — no concurrent 503
|
||||||
|
expect(a.bytes.equals(b.bytes)).toBe(true); // both got the same frame
|
||||||
|
});
|
||||||
|
|
||||||
|
it("reuses a fresh capture within the TTL (sequential, same vehicle)", async () => {
|
||||||
|
const { camera, calls } = fakeCamera();
|
||||||
|
const dev = id();
|
||||||
|
const a = await captureSnapshotShared(dev, camera, { direction: "entry" });
|
||||||
|
const b = await captureSnapshotShared(dev, camera, { direction: "entry" }); // ~0ms later
|
||||||
|
expect(calls()).toBe(1); // 2nd call served from the freshness cache
|
||||||
|
expect(a.bytes.equals(b.bytes)).toBe(true);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("pulls AGAIN after the TTL lapses (a later, different vehicle)", async () => {
|
||||||
|
vi.useFakeTimers();
|
||||||
|
try {
|
||||||
|
const { camera, calls } = fakeCamera();
|
||||||
|
const dev = id();
|
||||||
|
await captureSnapshotShared(dev, camera, { direction: "entry" });
|
||||||
|
expect(calls()).toBe(1);
|
||||||
|
await vi.advanceTimersByTimeAsync(2000); // past SNAPSHOT_TTL_MS (1500)
|
||||||
|
await captureSnapshotShared(dev, camera, { direction: "entry" });
|
||||||
|
expect(calls()).toBe(2); // stale → a real new pull (never a stale frame for a new car)
|
||||||
|
} finally {
|
||||||
|
vi.useRealTimers();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
it("does NOT cache a failure — the next caller retries", async () => {
|
||||||
|
const dev = id();
|
||||||
|
const failing = fakeCamera({ fail: true });
|
||||||
|
await expect(captureSnapshotShared(dev, failing.camera, { direction: "entry" })).rejects.toThrow("503");
|
||||||
|
// A subsequent capture (camera recovered) must actually pull, not inherit the error.
|
||||||
|
const ok = fakeCamera();
|
||||||
|
const shot = await captureSnapshotShared(dev, ok.camera, { direction: "entry" });
|
||||||
|
expect(shot.bytes.toString()).toBe("shot-x-1");
|
||||||
|
expect(ok.calls()).toBe(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("keys by deviceId — different cameras never share a frame", async () => {
|
||||||
|
const c1 = fakeCamera({ tag: "A" });
|
||||||
|
const c2 = fakeCamera({ tag: "B" });
|
||||||
|
const s1 = await captureSnapshotShared("cam-A", c1.camera, { direction: "entry" });
|
||||||
|
const s2 = await captureSnapshotShared("cam-B", c2.camera, { direction: "entry" });
|
||||||
|
expect(c1.calls()).toBe(1);
|
||||||
|
expect(c2.calls()).toBe(1);
|
||||||
|
expect(s1.bytes.equals(s2.bytes)).toBe(false);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -1,6 +1,6 @@
|
|||||||
import { randomUUID } from "node:crypto";
|
import { randomUUID } from "node:crypto";
|
||||||
import { deviceEvents as deviceEventsTable, snapshots, type Db } from "@parking/db";
|
import { deviceEvents as deviceEventsTable, snapshots, type Db } from "@parking/db";
|
||||||
import { registry, type CameraDevice } from "@parking/devices";
|
import { registry, type CameraDevice, type Snapshot } from "@parking/devices";
|
||||||
import type { FastifyBaseLogger } from "fastify";
|
import type { FastifyBaseLogger } from "fastify";
|
||||||
import { devicesByDirection, type FlowDirection } from "./device-resolve.js";
|
import { devicesByDirection, type FlowDirection } from "./device-resolve.js";
|
||||||
import type { VisionClient } from "./vision-client.js";
|
import type { VisionClient } from "./vision-client.js";
|
||||||
@@ -62,7 +62,9 @@ export function snapshotAsync(job: SnapshotJob): Promise<string[]> {
|
|||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
try {
|
try {
|
||||||
const shot = await camera.captureSnapshot({ direction });
|
// Shared capture: if the ANPR bridge just pulled this camera's frame for the
|
||||||
|
// same vehicle, reuse it instead of a 2nd concurrent GET (which 503s).
|
||||||
|
const shot = await captureSnapshotShared(row.id, camera, { direction });
|
||||||
const id: string = randomUUID();
|
const id: string = randomUUID();
|
||||||
db.insert(snapshots)
|
db.insert(snapshots)
|
||||||
.values({
|
.values({
|
||||||
@@ -153,6 +155,70 @@ export function buildCamera(row: { driverId: string; config: unknown }): CameraD
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// --- shared snapshot capture (one HTTP pull per camera per vehicle) -----------
|
||||||
|
// A Hikvision camera serves /ISAPI/.../picture SINGLE-THREADED: two concurrent
|
||||||
|
// snapshot GETs to the same unit return HTTP 503 "service busy". On a vehicle entry
|
||||||
|
// TWO paths capture the SAME camera within ~1s — the ANPR bridge (barrier-driving,
|
||||||
|
// anpr-entry.ts) and the advisory snapshotAsync (evidence + telemetry, below). They
|
||||||
|
// each `buildCamera()` a SEPARATE adapter instance, so a per-instance cache can't
|
||||||
|
// dedupe them. This module-level, deviceId-keyed cache does: it coalesces in-flight
|
||||||
|
// captures (the 2nd caller awaits the 1st's pull) AND serves a result captured within
|
||||||
|
// SNAPSHOT_TTL_MS, so the bridge + advisory share ONE frame instead of colliding into
|
||||||
|
// a 503 (which then burned the bridge's 12s debounce → the slow entry observed
|
||||||
|
// 2026-06-25; see wiki/concepts/lane-presence-and-anpr-entry.md).
|
||||||
|
|
||||||
|
/** How long a fresh capture is reused for the same camera. A car is one event for a
|
||||||
|
* couple of seconds; 1.5s comfortably spans the bridge→advisory gap without ever
|
||||||
|
* serving a stale frame for a *different* vehicle (entries are seconds apart). */
|
||||||
|
const SNAPSHOT_TTL_MS = 1500;
|
||||||
|
|
||||||
|
interface CacheEntry {
|
||||||
|
/** A capture in flight — concurrent callers await this instead of issuing a 2nd GET. */
|
||||||
|
inflight?: Promise<Snapshot>;
|
||||||
|
/** The last SUCCESSFUL capture + when it resolved, for the freshness window. */
|
||||||
|
last?: { shot: Snapshot; at: number };
|
||||||
|
}
|
||||||
|
|
||||||
|
const snapshotCache = new Map<string, CacheEntry>();
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Capture a snapshot for a camera, sharing ONE HTTP pull across concurrent/near-
|
||||||
|
* simultaneous callers (the ANPR bridge and the advisory snapshot). Same contract as
|
||||||
|
* `camera.captureSnapshot` (throws on failure) — a failed pull is NOT cached, so the
|
||||||
|
* next caller retries rather than inheriting the error. Key by the stable `deviceId`.
|
||||||
|
*/
|
||||||
|
export function captureSnapshotShared(
|
||||||
|
deviceId: string,
|
||||||
|
camera: CameraDevice,
|
||||||
|
ctx: { direction: FlowDirection },
|
||||||
|
): Promise<Snapshot> {
|
||||||
|
const now = Date.now();
|
||||||
|
let entry = snapshotCache.get(deviceId);
|
||||||
|
if (!entry) {
|
||||||
|
entry = {};
|
||||||
|
snapshotCache.set(deviceId, entry);
|
||||||
|
}
|
||||||
|
// Fresh enough → reuse the last frame (same vehicle, no second hardware hit).
|
||||||
|
if (entry.last && now - entry.last.at < SNAPSHOT_TTL_MS) {
|
||||||
|
return Promise.resolve(entry.last.shot);
|
||||||
|
}
|
||||||
|
// A capture is already running → join it (this is what prevents the 503 collision).
|
||||||
|
if (entry.inflight) return entry.inflight;
|
||||||
|
// Otherwise issue the single real pull; record it as the in-flight promise.
|
||||||
|
const pull = camera
|
||||||
|
.captureSnapshot(ctx)
|
||||||
|
.then((shot) => {
|
||||||
|
entry.last = { shot, at: Date.now() };
|
||||||
|
return shot;
|
||||||
|
})
|
||||||
|
.finally(() => {
|
||||||
|
// Clear the in-flight slot whether it resolved or threw; a failure is never cached.
|
||||||
|
if (entry.inflight === pull) entry.inflight = undefined;
|
||||||
|
});
|
||||||
|
entry.inflight = pull;
|
||||||
|
return pull;
|
||||||
|
}
|
||||||
|
|
||||||
function recordFailure(
|
function recordFailure(
|
||||||
db: Db,
|
db: Db,
|
||||||
direction: FlowDirection,
|
direction: FlowDirection,
|
||||||
|
|||||||
Reference in New Issue
Block a user