fix(gps): остановлен 429-шторм — throttle SSE gps_asset_moved (3 сек/asset) и клиентских инвалидаций (60 сек), polling позиций 60 сек, статусов 5 мин

Каждая точка ingest (~1/сек) broadcast'илась SSE всем клиентам орг., клиент
на каждое событие делал invalidateQueries без throttle → 2+ GET/сек → исчерпание
per-user бакета apiLimiter (1500/15мин) → все API устройства 429 (формы не грузились).
Модель теперь как у нормальных трекеров: точки пишутся в БД, карта читает по интервалу.
+4 unit-теста троттла (server/gps/sse-throttle.ts)
This commit is contained in:
2026-09-17 21:59:34 +03:00
parent e4f9a1992a
commit c1d5b1cbee
4 changed files with 133 additions and 33 deletions

View File

@@ -1,4 +1,4 @@
import { useCallback } from 'react';
import { useCallback, useEffect, useRef } from 'react';
import { useQuery } from '@tanstack/react-query';
import { apiRequest, queryClient } from '@/lib/queryClient';
import { useEvents } from '@/hooks/useEvents';
@@ -25,29 +25,64 @@ export function useGpsConfig() {
export function useGpsAssets() {
return useQuery<{ success: boolean; assets: GpsAsset[] }>({
queryKey: ['/api/gps/assets'],
refetchInterval: 30_000, // автообновление статусов онлайн/«Нет GPS»
refetchInterval: 300_000, // автообновление статусов онлайн/«Нет GPS» — не критично к минутной задержке
});
}
/**
* Текущие позиции активных объектов.
* Обновляются по SSE (gps_asset_moved / gps_asset_offline) + fallback polling 30 сек.
* Обновляются по SSE (gps_asset_moved / gps_asset_offline, серверный throttle 3 сек)
* + fallback polling 60 сек.
*/
export function useGpsPositions(enabled = true) {
return useQuery<{ success: boolean; positions: GpsPosition[] }>({
queryKey: ['/api/gps/assets/positions'],
refetchInterval: 30_000,
refetchInterval: 60_000,
enabled,
});
}
/** Подписка на живые GPS-события: инвалидирует позиции и список объектов */
/** Минимальный интервал между инвалидациями по SSE — иначе поток событий исчерпывает apiLimiter (429) */
const GPS_LIVE_INVALIDATE_INTERVAL_MS = 60_000;
/** Подписка на живые GPS-события: инвалидирует позиции и список объектов (trailing throttle 60 сек) */
export function useGpsLiveUpdates(enabled = true) {
const invalidate = useCallback(() => {
const timerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
const pendingRef = useRef(false);
const invalidateNow = useCallback(() => {
queryClient.invalidateQueries({ queryKey: ['/api/gps/assets/positions'] });
queryClient.invalidateQueries({ queryKey: ['/api/gps/assets'] });
}, []);
const invalidate = useCallback(() => {
// Окно ещё не открыто — инвалидируем сразу и открываем окно
if (timerRef.current === null) {
invalidateNow();
timerRef.current = setTimeout(() => {
timerRef.current = null;
// Trailing: события, накопившиеся за окно, дают одну инвалидацию по выходу
if (pendingRef.current) {
pendingRef.current = false;
invalidateNow();
}
}, GPS_LIVE_INVALIDATE_INTERVAL_MS);
} else {
pendingRef.current = true;
}
}, [invalidateNow]);
useEffect(
() => () => {
if (timerRef.current !== null) {
clearTimeout(timerRef.current);
timerRef.current = null;
}
pendingRef.current = false;
},
[]
);
useEvents(
enabled
? {

View File

@@ -1,5 +1,6 @@
import { gpsStorage } from "./storage";
import { OSRM_URL } from "./osrm";
import { shouldPublishGpsAssetMoved } from "./sse-throttle";
import { storage } from "../storage";
import { webPushService } from "../services/web-push.service";
import { eventBus, publishNotificationSSE } from "../routes/shared";
@@ -521,19 +522,22 @@ export async function processPosition(
recordedAt,
isLbs: true,
});
eventBus.publishEvent({
type: "gps_asset_moved",
data: {
assetId,
lat: input.lat,
lng: input.lng,
speed: input.speed ?? null,
course: input.course ?? null,
recordedAt: recordedAt.toISOString(),
isLbs: true,
},
organizationId,
});
// SSE-broadcast троттлим (не чаще раза в 3 сек на объект) — запись в БД выше не трогаем
if (shouldPublishGpsAssetMoved(assetId)) {
eventBus.publishEvent({
type: "gps_asset_moved",
data: {
assetId,
lat: input.lat,
lng: input.lng,
speed: input.speed ?? null,
course: input.course ?? null,
recordedAt: recordedAt.toISOString(),
isLbs: true,
},
organizationId,
});
}
return;
}
@@ -633,20 +637,24 @@ export async function processPosition(
}
// 3. SSE-событие о перемещении — broadcast по организации,
// чтобы карта у всех открытых клиентов обновилась в реальном времени
eventBus.publishEvent({
type: "gps_asset_moved",
data: {
assetId,
lat: input.lat,
lng: input.lng,
speed: input.speed ?? null,
course: input.course ?? null,
recordedAt: recordedAt.toISOString(),
isLbs,
},
organizationId,
});
// чтобы карта у всех открытых клиентов обновилась в реальном времени.
// Троттлим не чаще раза в 3 сек на объект, иначе при потоке точек
// ~1/сек клиентская инвалидация исчерпывает apiLimiter (429).
if (shouldPublishGpsAssetMoved(assetId)) {
eventBus.publishEvent({
type: "gps_asset_moved",
data: {
assetId,
lat: input.lat,
lng: input.lng,
speed: input.speed ?? null,
course: input.course ?? null,
recordedAt: recordedAt.toISOString(),
isLbs,
},
organizationId,
});
}
// 4. Проверка геозон — только для настоящих GPS-точек;
// LBS-координаты вышек в логику геозон не пускаем

View File

@@ -0,0 +1,27 @@
/**
* In-memory throttle для SSE-событий gps_asset_moved.
* Трекеры присылают точки ~1 раз в секунду; без ограничения каждый открытый
* клиент организации получает broadcast каждую секунду и инвалидирует запросы
* позиций/объектов, что исчерпывает per-user бакет apiLimiter (429).
* Запись позиции в БД не затрагивается — ограничивается только broadcast.
*/
const GPS_MOVED_SSE_MIN_INTERVAL_MS = 3_000;
const MAX_TRACKED_ASSETS = 1_000;
const lastPublishedAt = new Map<number, number>();
/** true — событие публиковать; false — пропустить (с последней публикации прошло меньше интервала) */
export function shouldPublishGpsAssetMoved(assetId: number, now = Date.now()): boolean {
const last = lastPublishedAt.get(assetId);
if (last !== undefined && now - last < GPS_MOVED_SSE_MIN_INTERVAL_MS) return false;
lastPublishedAt.set(assetId, now);
// Защита от неограниченного роста карты: чистим устаревшие записи при превышении лимита
if (lastPublishedAt.size > MAX_TRACKED_ASSETS) {
for (const [id, ts] of lastPublishedAt) {
if (now - ts >= GPS_MOVED_SSE_MIN_INTERVAL_MS) lastPublishedAt.delete(id);
}
}
return true;
}

View File

@@ -0,0 +1,30 @@
import { describe, it, expect } from 'vitest';
import { shouldPublishGpsAssetMoved } from '../server/gps/sse-throttle';
describe('shouldPublishGpsAssetMoved', () => {
it('первое событие объекта публикуется сразу', () => {
expect(shouldPublishGpsAssetMoved(1, 10_000)).toBe(true);
});
it('события чаще 3 секунд подряд пропускаются', () => {
const assetId = 2;
expect(shouldPublishGpsAssetMoved(assetId, 100_000)).toBe(true);
expect(shouldPublishGpsAssetMoved(assetId, 101_000)).toBe(false);
expect(shouldPublishGpsAssetMoved(assetId, 102_999)).toBe(false);
});
it('после окна в 3 секунды событие снова публикуется', () => {
const assetId = 3;
expect(shouldPublishGpsAssetMoved(assetId, 200_000)).toBe(true);
expect(shouldPublishGpsAssetMoved(assetId, 203_000)).toBe(true);
expect(shouldPublishGpsAssetMoved(assetId, 204_000)).toBe(false);
});
it('throttle независим между объектами', () => {
const now = 300_000;
expect(shouldPublishGpsAssetMoved(4, now)).toBe(true);
expect(shouldPublishGpsAssetMoved(5, now)).toBe(true);
expect(shouldPublishGpsAssetMoved(4, now + 1_000)).toBe(false);
expect(shouldPublishGpsAssetMoved(5, now + 1_000)).toBe(false);
});
});