From b3cb67188eecbd54c7e1bd5fd8abc6a5a798f837 Mon Sep 17 00:00:00 2001 From: Julian Cuni Date: Fri, 26 Jun 2026 08:10:56 +0200 Subject: [PATCH] fix(anpr): share one camera snapshot across bridge + advisory paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- apps/server/src/anpr-entry.test.ts | 5 ++ apps/server/src/anpr-entry.ts | 6 +- apps/server/src/snapshot.test.ts | 93 ++++++++++++++++++++++++++++++ apps/server/src/snapshot.ts | 70 +++++++++++++++++++++- 4 files changed, 170 insertions(+), 4 deletions(-) create mode 100644 apps/server/src/snapshot.test.ts diff --git a/apps/server/src/anpr-entry.test.ts b/apps/server/src/anpr-entry.test.ts index e921e2d..21307b0 100644 --- a/apps/server/src/anpr-entry.test.ts +++ b/apps/server/src/anpr-entry.test.ts @@ -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 // (no registry, no network). The factory returns a fresh shot each call. 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", () => ({ buildCamera: () => ({ captureSnapshot }), + captureSnapshotShared: (_id: string, camera: { captureSnapshot: typeof captureSnapshot }, ctx: unknown) => + camera.captureSnapshot(ctx as never), })); // Import AFTER the mock is registered. diff --git a/apps/server/src/anpr-entry.ts b/apps/server/src/anpr-entry.ts index f669388..e57e57b 100644 --- a/apps/server/src/anpr-entry.ts +++ b/apps/server/src/anpr-entry.ts @@ -3,7 +3,7 @@ import { devices, deviceEvents as deviceEventsTable, eq, siteConfig, type Db, ty import type { FastifyBaseLogger } from "fastify"; import { deviceEvents, type DeviceReadEvent } from "./device-events.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 { 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 — // the gated flow infers the verb from the camera's bound relay direction). 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); if (!result || !result.plate) return; // nothing read diff --git a/apps/server/src/snapshot.test.ts b/apps/server/src/snapshot.test.ts new file mode 100644 index 0000000..16d50ee --- /dev/null +++ b/apps/server/src/snapshot.test.ts @@ -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 { + 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); + }); +}); diff --git a/apps/server/src/snapshot.ts b/apps/server/src/snapshot.ts index 9de4d61..3092f82 100644 --- a/apps/server/src/snapshot.ts +++ b/apps/server/src/snapshot.ts @@ -1,6 +1,6 @@ import { randomUUID } from "node:crypto"; 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 { devicesByDirection, type FlowDirection } from "./device-resolve.js"; import type { VisionClient } from "./vision-client.js"; @@ -62,7 +62,9 @@ export function snapshotAsync(job: SnapshotJob): Promise { return null; } 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(); db.insert(snapshots) .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; + /** The last SUCCESSFUL capture + when it resolved, for the freshness window. */ + last?: { shot: Snapshot; at: number }; +} + +const snapshotCache = new Map(); + +/** + * 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 { + 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( db: Db, direction: FlowDirection,