diff --git a/backend/src/modules/mcp/action-types.ts b/backend/src/modules/mcp/action-types.ts index 1c7fc291..a25fa4a0 100644 --- a/backend/src/modules/mcp/action-types.ts +++ b/backend/src/modules/mcp/action-types.ts @@ -3,11 +3,14 @@ import type { PoolClient } from "pg" import type { McpOperationDescriptor, McpPolicyPreviewRequest } from "@mmapp/contracts/mcp" import type { ConnectionRow, McpPrincipal } from "./store.js" import type { RosSnapshot, RosStep } from "./resources.js" +import type { RoutingConfirmationObservations } from "./routing-model.js" export interface ActionContext { connection: ConnectionRow; principal: McpPrincipal; signal?: AbortSignal; logsFollowMaintenance?: AbortSignal /** Process-internal persistent controller identity; never parsed from tool JSON. */ persistentRoutingProfile?: { id: string; version: number } + /** Process-internal sink for the controller's measurement identities. */ + captureRoutingConfirmation?: (observations: RoutingConfirmationObservations) => void /** Available only during target resolution, never retained by the read/prepare handler. */ transactionClient?: Pick } diff --git a/backend/src/modules/mcp/actions.ts b/backend/src/modules/mcp/actions.ts index ca463f43..6b5639d0 100644 --- a/backend/src/modules/mcp/actions.ts +++ b/backend/src/modules/mcp/actions.ts @@ -422,11 +422,17 @@ add("logs.query", "mm_logs_query", "История логов", "logs", "mm:serv add("logs.summary", "mm_logs_summary_get", "Статистика логов", "logs", "mm:servers:read", "read", "servers", logInput, { read: async (input, context) => { const { settings: _settings, ...summary } = await readLogSummary(logFilters(input, context)); return summary }, }) -for (const context of [false, true]) add(context ? "logs.context" : "logs.get", context ? "mm_logs_context_get" : "mm_logs_get", "Запись журнала", "logs", "mm:servers:read", "read", "servers", z.object({ serverId, id: z.string().regex(/^\d{1,20}$/) }).strict(), { +const maxLogEntryId = "9223372036854775807" +const logEntryId = z.string().regex(/^\d{1,20}$/).refine(id => { + const digits = id.replace(/^0+/, "") + return digits.length < maxLogEntryId.length || (digits.length === maxLogEntryId.length && digits <= maxLogEntryId) +}, "ID журнала выходит за допустимый диапазон") +for (const withContext of [false, true]) add(withContext ? "logs.context" : "logs.get", withContext ? "mm_logs_context_get" : "mm_logs_get", "Запись журнала", "logs", "mm:servers:read", "read", "servers", z.object({ serverId, id: logEntryId }).strict(), { read: async input => { - const entry = await readLogEntry(String(input.id)) - if (!entry || entry.serverId !== input.serverId) throw new McpError("NOT_FOUND", "Событие не найдено", 404) - return context ? readLogContext(String(input.id)) : entry + const id = String(input.id), serverId = Number(input.serverId) + const result = await (withContext ? readLogContext(id, serverId) : readLogEntry(id, serverId)) + if (!result) throw new McpError("NOT_FOUND", "Событие не найдено", 404) + return result }, }) add("logs.follow", "mm_logs_follow", "Новые события", "logs", "mm:servers:read", "read", "servers", logInput.safeExtend({ serverId, limit: logInput.shape.limit.default(200), cursor: z.string().max(8192).optional(), waitSeconds: z.number().int().min(0).max(30).default(30) }), { diff --git a/backend/src/modules/mcp/routing-actions.ts b/backend/src/modules/mcp/routing-actions.ts index e08ff20a..153a4c71 100644 --- a/backend/src/modules/mcp/routing-actions.ts +++ b/backend/src/modules/mcp/routing-actions.ts @@ -61,7 +61,9 @@ async function analyzeCommunityProfile(profile: z.infer Number(b.eligible) - Number(a.eligible)) - return { profile, ranking, ...routeContext.before, observedPath: false as const, returnPathVerified: false as const, observedAt: new Date().toISOString() } + context.captureRoutingConfirmation?.(new Map(ranking.map(row => [row.candidate.id, row.confirmationObservationIds]))) + const publicRanking = ranking.map(({ confirmationObservationIds: _private, ...row }) => row) + return { profile, ranking: publicRanking, ...routeContext.before, observedPath: false as const, returnPathVerified: false as const, observedAt: new Date().toISOString() } } async function assertCandidateEligible(profile: z.infer, candidateId: string) { diff --git a/backend/src/modules/mcp/routing-model.test.ts b/backend/src/modules/mcp/routing-model.test.ts index c0acd20e..4fbe3709 100644 --- a/backend/src/modules/mcp/routing-model.test.ts +++ b/backend/src/modules/mcp/routing-model.test.ts @@ -142,6 +142,50 @@ test("controller does not count polling the same sample as consecutive confirmat assert.equal(decideRoutingSwitch(profile, state, ranks, null, now).switchTo, null) }) +test("controller counts only samples used for ranking or SLA and resets an old mixed streak", () => { + const slow = a.map(row => ({ ...row, pingMs: 100, bandwidthAt: null, dlMbps: null, ulMbps: null })) + const base = [...slow, sample("b1", 1, 4, "wan-b", { pingMs: 1, bandwidthAt: null, dlMbps: null, ulMbps: null }), + sample("b2", 4, 3, "exit-b", { pingMs: 1, bandwidthAt: null, dlMbps: null, ulMbps: null })] + const bandwidth = base.map(row => ({ ...row, id: `bw-${row.id}`, pingAt: null, pingMs: null, lossPct: null, + bandwidthAt: at, dlMbps: 100, ulMbps: 100 })) + const pingOnly = routeAiProfileSchema.parse({ ...profile, weights: { ping: 100, speed: 0, loss: 0 } }) + const decide = (configured: typeof profile, state: ControllerState, observations: RouteObservation[]) => { + const ranks = rankRoutingCandidates(configured, observations, now + 20) + const confirmation = new Map(ranks.map(row => [row.candidate.id, row.confirmationObservationIds])) + return decideRoutingSwitch(configured, state, ranks, "a", now + 20, [], undefined, confirmation) + } + let state: ControllerState = { pendingCandidateId: "b", consecutive: 2, observationKey: "old-mixed-key" } + for (let i = 0; i < 3; i++) { + const observations = [...base, ...bandwidth, { ...bandwidth[2], id: `bw-new-${i}`, bandwidthAt: new Date(now + i + 1).toISOString() }] + const result = decide(pingOnly, state, observations) + assert.equal(result.switchTo, null) + assert.equal(result.state.consecutive, 1) + state = result.state + } + assert.equal(state.observationKey, "ping:a1:a2:b1:b2") + const fresh = (i: number) => [...base, ...bandwidth, { ...base[2], id: `ping-new-${i}`, pingAt: new Date(now + i + 1).toISOString() }] + const second = decide(pingOnly, state, fresh(1)) + assert.equal(second.state.consecutive, 2) + assert.equal(decide(pingOnly, second.state, fresh(1)).state.consecutive, 2) + assert.equal(decide(pingOnly, second.state, fresh(2)).switchTo, "b") + + const withBandwidthSla = routeAiProfileSchema.parse({ ...pingOnly, minBandwidthMbps: 10 }) + const initial = decide(withBandwidthSla, {}, [...base, ...bandwidth]) + const updated = decide(withBandwidthSla, initial.state, [...base, ...bandwidth, { + ...bandwidth[2], id: "new-sla-sample", bandwidthAt: new Date(now + 1).toISOString(), + }]) + assert.equal(initial.state.observationKey, "a1:a2:b1:b2:bw-a1:bw-a2:bw-b1:bw-b2") + assert.equal(updated.state.consecutive, 2) + + const bandwidthOnly = routeAiProfileSchema.parse({ ...profile, weights: { ping: 0, speed: 100, loss: 0 } }) + const unequalBandwidth = bandwidth.map((row, i) => ({ ...row, dlMbps: i < 2 ? 20 : 800, ulMbps: i < 2 ? 20 : 800 })) + const speedInitial = decide(bandwidthOnly, {}, [...base, ...unequalBandwidth]) + const speedWithNewPing = decide(bandwidthOnly, speedInitial.state, [...base, ...unequalBandwidth, fresh(1).at(-1)!]) + assert.equal(speedInitial.state.observationKey, "bandwidth:bw-a1:bw-a2:bw-b1:bw-b2") + assert.equal(speedWithNewPing.state.consecutive, 1) + assert.equal(speedWithNewPing.switchTo, null) +}) + test("hysteresis retains pending confirmation without counting sub-threshold observations or releasing cooldown", () => { const ranked = rankRoutingCandidates(profile, [...a, sample("b1", 1, 4, "wan-b"), sample("b2", 4, 3, "exit-b")], now) const compare = (state: ControllerState, gain: number, observation: string, time = now) => { diff --git a/backend/src/modules/mcp/routing-model.ts b/backend/src/modules/mcp/routing-model.ts index 0a646dca..61116750 100644 --- a/backend/src/modules/mcp/routing-model.ts +++ b/backend/src/modules/mcp/routing-model.ts @@ -4,7 +4,14 @@ export function fresh(timestamp: string | null, seconds: number, now: number): b const measured = timestamp === null ? NaN : Date.parse(timestamp) return Number.isFinite(measured) && measured <= now && now - measured <= seconds * 1000 } +function confirmationMetrics(profile: RouteAiProfile) { + return { + ping: profile.weights.ping > 0 || profile.weights.loss > 0 || profile.maxPingMs !== null || profile.maxLossPct !== null, + bandwidth: profile.weights.speed > 0 || profile.minBandwidthMbps !== null, + } +} export function rankRoutingCandidates(profile: RouteAiProfile, observations: RouteObservation[], now = Date.now()) { + const metrics = confirmationMetrics(profile) function leg(src: number, dst: number, iface: string, remoteIface?: string) { const possible = observations.flatMap(observation => { if (observation.family !== profile.family || observation.table !== profile.table) return [] @@ -22,7 +29,8 @@ export function rankRoutingCandidates(profile: RouteAiProfile, observations: Rou const ping = latest("pingAt"), bandwidth = latest("bandwidthAt"), base = ping ?? bandwidth return base ? { ...base, pingMs: ping?.contextVerified ? ping.pingMs : null, lossPct: ping?.contextVerified ? ping.lossPct : null, pingAt: ping?.pingAt ?? null, dlMbps: bandwidth?.contextVerified ? bandwidth.dlMbps : null, ulMbps: bandwidth?.contextVerified ? bandwidth.ulMbps : null, bandwidthAt: bandwidth?.bandwidthAt ?? null, - sampleIds: [...new Set([ping?.id, bandwidth?.id].filter((id): id is string => !!id))] } : undefined + sampleIds: [...new Set([ping?.id, bandwidth?.id].filter((id): id is string => !!id))], + confirmationIds: [metrics.ping ? ping?.id : undefined, metrics.bandwidth ? bandwidth?.id : undefined].filter((id): id is string => !!id) } : undefined } return profile.candidates.map(candidate => { const first = leg(profile.serverId, candidate.jumpHostId, candidate.wanInterface, candidate.wanRemoteInterface) @@ -46,14 +54,18 @@ export function rankRoutingCandidates(profile: RouteAiProfile, observations: Rou if (lossPct !== null && profile.maxLossPct !== null && lossPct > profile.maxLossPct) reasons.push("Превышены потери") if (dlMbps !== null && ulMbps !== null && profile.minBandwidthMbps !== null && Math.min(dlMbps, ulMbps) < profile.minBandwidthMbps) reasons.push("Недостаточна пропускная способность") const score = reasons.length ? null : Math.round((profile.weights.ping * Math.max(0, 100 - (pingMs ?? 0) * 0.6) + profile.weights.speed * Math.min(100, ((dlMbps ?? 0) + (ulMbps ?? 0)) / 18) + profile.weights.loss * (100 - (lossPct ?? 0))) / 100) - return { candidate, eligible: !reasons.length, score, reasons, pingMs, lossPct, dlMbps, ulMbps, observationIds: legs.flatMap(value => value?.sampleIds ?? []), observedPath: false as const } + return { candidate, eligible: !reasons.length, score, reasons, pingMs, lossPct, dlMbps, ulMbps, observationIds: legs.flatMap(value => value?.sampleIds ?? []), + confirmationObservationIds: legs.flatMap(value => value?.confirmationIds ?? []), observedPath: false as const } }).sort((a, b) => (b.score ?? -1) - (a.score ?? -1) || a.candidate.id.localeCompare(b.candidate.id)) } export interface PathHealth { id: string; candidateId: string; status: "passed" | "failed" | "unknown"; observedAt: string } export interface ControllerState { pendingCandidateId?: string; consecutive?: number; observationKey?: string; lastSwitchAt?: string; switches?: string[]; operationId?: string; pausedReason?: string; failureCount?: number; failureKey?: string; failureObservedAt?: string; failingCandidateId?: string } -export function selectFailoverReserveProbes(ranking: ReturnType, currentCandidateId: string, previous?: string) { +export type PublicRoutingRanking = Array[number], "confirmationObservationIds">> +export type RoutingConfirmationObservations = ReadonlyMap + +export function selectFailoverReserveProbes(ranking: PublicRoutingRanking, currentCandidateId: string, previous?: string) { const standbys = ranking.filter(({ candidate, eligible }) => eligible && candidate.id !== currentCandidateId && candidate.controlProbe) .map(({ candidate }) => candidate.id) const [preferred, ...remaining] = standbys @@ -64,7 +76,7 @@ export function selectFailoverReserveProbes(ranking: ReturnType, currentCandidateId: string | null, now = Date.now(), health: PathHealth[] = [], chargedSwitchesPerHour?: number) { +export function decideRoutingSwitch(profile: RouteAiProfile, state: ControllerState, ranking: PublicRoutingRanking, currentCandidateId: string | null, now = Date.now(), health: PathHealth[] = [], chargedSwitchesPerHour?: number, confirmation?: RoutingConfirmationObservations) { const next: ControllerState = { ...state, switches: (state.switches ?? []).filter(at => now - Date.parse(at) < 3600000) } if (chargedSwitchesPerHour !== undefined && (!Number.isSafeInteger(chargedSwitchesPerHour) || chargedSwitchesPerHour < next.switches!.length || chargedSwitchesPerHour > 4096)) throw Error("Invalid common routing charge count") const switchCount = chargedSwitchesPerHour ?? next.switches!.length @@ -93,7 +105,18 @@ export function decideRoutingSwitch(profile: RouteAiProfile, state: ControllerSt const gain = best.score! - current.score if (gain < Math.max(0, profile.switchThreshold - profile.hysteresis)) return { state: { ...next, consecutive: 0, pendingCandidateId: undefined }, switchTo: null, reason: "Сработал гистерезис" } if (gain < profile.switchThreshold) return { state: next, switchTo: null, reason: "Порог улучшения не достигнут" } - const observationKey = [...best.observationIds, ...current.observationIds].sort().join(":") + const metrics = confirmationMetrics(profile) + let observationKey: string + if (metrics.ping && metrics.bandwidth) { + observationKey = [...best.observationIds, ...current.observationIds].sort().join(":") + } else { + const bestIds = confirmation?.get(best.candidate.id), currentIds = confirmation?.get(current.candidate.id) + if (!bestIds?.length || !currentIds?.length) return { state: next, switchTo: null, reason: "Нет подтверждённых измерений для выбора" } + const prefix = metrics.ping ? "ping:" : "bandwidth:" + observationKey = prefix + [...new Set([...bestIds, ...currentIds])].sort().join(":") + // Old streaks may include samples irrelevant to the current policy. + if (!next.observationKey?.startsWith(prefix)) next.consecutive = 0 + } if (next.pendingCandidateId !== undefined && next.pendingCandidateId !== best.candidate.id) { next.pendingCandidateId = best.candidate.id next.consecutive = 0 diff --git a/backend/src/modules/mcp/routing-profiles.ts b/backend/src/modules/mcp/routing-profiles.ts index ba8a0fd0..1dfa34c6 100644 --- a/backend/src/modules/mcp/routing-profiles.ts +++ b/backend/src/modules/mcp/routing-profiles.ts @@ -9,7 +9,7 @@ import { pool } from "../../db/index.js" import { registerMcpService } from "./service-registry.js" import { accessGeneration, McpError } from "./security.js" import { audit, validatePrincipal, type McpPrincipal } from "./store.js" -import { decideRoutingSwitch, selectFailoverReserveProbes, type ControllerState, type PathHealth, type rankRoutingCandidates } from "./routing-model.js" +import { decideRoutingSwitch, selectFailoverReserveProbes, type ControllerState, type PathHealth, type PublicRoutingRanking, type RoutingConfirmationObservations } from "./routing-model.js" import type { add } from "./actions.js" import type { ActionContext } from "./action-types.js" import { COMMUNITY_PATH_MAX_AGE_MS } from "./path-proof-purpose.js" @@ -124,7 +124,10 @@ export async function tickRoutingProfiles() { await assertRoutingControllerBaseline(state.baseline,row.id,row.version,profile,row.grant_expires_at.toISOString(),{principal,connection}) if (state.probedAt && Date.now() - Date.parse(state.probedAt) < 55000) return await current() - const analysis = await invokeMcpTool(principal, "mm_route_optimizer_analyze", { profile, limit: 100 }) as { ranking: ReturnType; currentCandidateId: string | null } + let confirmation: RoutingConfirmationObservations | undefined + const analysis = await invokeMcpTool(principal, "mm_route_optimizer_analyze", { profile, limit: 100 }, undefined, { + captureRoutingConfirmation: observations => { confirmation = observations }, + }) as { ranking: PublicRoutingRanking; currentCandidateId: string | null } if (state.expectedCandidateId && analysis.currentCandidateId !== state.expectedCandidateId) { await stop("Наблюдаемый путь изменился вне ожидаемого переключения; требуется проверка вмешательства или дрейфа"); return } if (analysis.currentCandidateId) next.expectedCandidateId = analysis.currentCandidateId if (profile.candidates.some(candidate => candidate.controlProbe) && !connection.policy.scope.operationIds.includes("routing.path.verify")) { await stop("Не выдано право контрольной проверки пути"); return } @@ -152,7 +155,7 @@ export async function tickRoutingProfiles() { const age = decisionAt - Date.parse(probe.observedAt) return age >= 0 && age < COMMUNITY_PATH_MAX_AGE_MS }) - const decision = decideRoutingSwitch(profile, state, analysis.ranking, analysis.currentCandidateId, decisionAt, health, charges.totalCharges) + const decision = decideRoutingSwitch(profile, state, analysis.ranking, analysis.currentCandidateId, decisionAt, health, charges.totalCharges, confirmation) Object.assign(next, decision.state, { reason: decision.reason }) if (!failoverProbes?.cursor) delete next.failoverReserveCursor else if (!hasFreshPassedProbe(failoverProbes.cursor)) { diff --git a/backend/src/modules/mcp/runtime.ts b/backend/src/modules/mcp/runtime.ts index f98b1572..52f6fb45 100644 --- a/backend/src/modules/mcp/runtime.ts +++ b/backend/src/modules/mcp/runtime.ts @@ -125,7 +125,10 @@ async function completeRequest(id: string, started: number, result: unknown, err await assertDeliveryAccess?.(client) }) } -export async function invokeMcpTool(principal: McpPrincipal, toolName: string, raw: unknown, signal?: AbortSignal, internal?: { persistentRoutingProfile: { id: string; version: number } }): Promise { +type InternalMcpCall = { persistentRoutingProfile: NonNullable } + | { captureRoutingConfirmation: NonNullable } + +export async function invokeMcpTool(principal: McpPrincipal, toolName: string, raw: unknown, signal?: AbortSignal, internal?: InternalMcpCall): Promise { if (toolName === "mm_logs_follow") { const maintenance = new AbortController() return sharedDatabaseGate.request( @@ -135,19 +138,24 @@ export async function invokeMcpTool(principal: McpPrincipal, toolName: string, r } return sharedDatabaseGate.request(() => invokeMcpToolInner(principal, toolName, raw, signal, internal)) } -async function invokeMcpToolInner(principal: McpPrincipal, toolName: string, raw: unknown, signal?: AbortSignal, internal?: { persistentRoutingProfile: { id: string; version: number } }, logsFollowMaintenance?: AbortSignal): Promise { +async function invokeMcpToolInner(principal: McpPrincipal, toolName: string, raw: unknown, signal?: AbortSignal, internal?: InternalMcpCall, logsFollowMaintenance?: AbortSignal): Promise { const action = findAction(toolName), input = action.schema.parse(raw) as Record, started = Date.now() const call = await reserveCall(principal, toolName, input, signal) const cancelled = () => logsFollowMaintenance && logsFollowCancellation(logsFollowMaintenance, signal) const assertActive = () => { const failure = cancelled(); if (failure) throw failure } let vrfPreparing = false + let routingConfirmation: Parameters>[0] | undefined try { assertActive() if (logsFollowMaintenance) call.context.logsFollowMaintenance = logsFollowMaintenance - if (internal) { + if (internal && "persistentRoutingProfile" in internal) { if (action.operation.id !== "routing.community.switch") throw new McpError("INVALID_INPUT", "Контекст профиля не относится к этой операции", 400) call.context.persistentRoutingProfile = z.object({ id: z.string().uuid(), version: z.number().int().positive().safe() }).strict().parse(internal.persistentRoutingProfile) } + if (internal && "captureRoutingConfirmation" in internal) { + if (action.operation.id !== "routing.analyze") throw new McpError("INVALID_INPUT", "Измерения контроллера не относятся к этой операции", 400) + call.context.captureRoutingConfirmation = observations => { routingConfirmation = observations } + } let result: unknown if (action.read) result = await action.read(input, call.context) else if (action.prepare) { @@ -213,6 +221,7 @@ async function invokeMcpToolInner(principal: McpPrincipal, toolName: string, raw await completeRequest(call.requestId, started, output, undefined, checkDelivery) await checkDelivery?.() assertActive() + if (routingConfirmation && internal && "captureRoutingConfirmation" in internal) internal.captureRoutingConfirmation(routingConfirmation) return output } catch (error) { if (vrfPreparing) { diff --git a/backend/src/services/router-logs-model.ts b/backend/src/services/router-logs-model.ts index ec90155a..5d6b43cf 100644 --- a/backend/src/services/router-logs-model.ts +++ b/backend/src/services/router-logs-model.ts @@ -31,11 +31,19 @@ export function parseLogTime(raw: string): string | null { const ms = Date.parse(raw.replace(" ", "T")) return Number.isFinite(ms) ? new Date(ms).toISOString() : null } -export function logWhere(filters: LogFilters & { allowedServerIds?: number[] }, extra: string[] = [], initial: unknown[] = []) { +type LogServerScope = Pick & { allowedServerIds?: number[] } + +export function logServerConditions({ allowedServerIds, serverId }: LogServerScope, bind: (value: unknown) => string) { + const clauses: string[] = [] + if (allowedServerIds !== undefined) clauses.push(`server_id = ANY(${bind(allowedServerIds)}::bigint[])`) + if (serverId) clauses.push(`server_id = ${bind(serverId)}`) + return clauses +} + +export function logWhere(filters: LogFilters & LogServerScope, extra: string[] = [], initial: unknown[] = []) { const values = [...initial], clauses = [...extra] const bind = (value: unknown) => { values.push(value); return `$${values.length}` } - if (filters.allowedServerIds !== undefined) clauses.push(`server_id = ANY(${bind(filters.allowedServerIds)}::bigint[])`) - if (filters.serverId) clauses.push(`server_id = ${bind(filters.serverId)}`) + clauses.push(...logServerConditions(filters, bind)) if (filters.from) clauses.push(`effective_at >= ${bind(filters.from)}::timestamptz`) if (filters.to) clauses.push(`effective_at <= ${bind(filters.to)}::timestamptz`) if (filters.levels) clauses.push(`level = ANY(${bind(filters.levels.split(","))}::text[])`) diff --git a/backend/src/services/router-logs-query.ts b/backend/src/services/router-logs-query.ts index 0d8a8184..fb5844cd 100644 --- a/backend/src/services/router-logs-query.ts +++ b/backend/src/services/router-logs-query.ts @@ -2,7 +2,7 @@ import type { PoolClient } from "pg" import { logLevels, type LogEntry, type LogFilters } from "@mmapp/contracts" import { pool } from "../db/index.js" import { getLogSettings, getLogStates } from "./router-logs-collector.js" -import { logWhere } from "./router-logs-model.js" +import { logServerConditions, logWhere } from "./router-logs-model.js" export const logColumns = `id::text, server_id AS "serverId",server_name AS "serverName",host,router_id AS "routerId",raw_time AS "rawTime",event_at AS "eventAt",received_at AS "receivedAt",effective_at AS "effectiveAt",CASE WHEN event_at IS NULL THEN 'receipt' ELSE 'router' END AS "timeKind",topics,level,message,generation` export function encodeLogCursor(row: LogEntry) { return Buffer.from(JSON.stringify([row.effectiveAt, row.id])).toString("base64url") } @@ -13,15 +13,18 @@ export function decodeLogCursor(cursor: string): [string, string] { } catch { /* invalid cursor */ } throw Object.assign(new Error("Некорректный курсор журнала"), { statusCode: 400 }) } -export async function logWatermark(client: PoolClient, serverId?: number) { - const res = await client.query<{ id: string }>(`SELECT COALESCE(max(id),0)::text AS id FROM router_log_entries${serverId ? " WHERE server_id=$1" : ""}`, serverId ? [serverId] : []) +export async function logWatermark(client: PoolClient, serverId?: number, allowedServerIds?: number[]) { + const values: unknown[] = [] + const clauses = logServerConditions({ serverId, allowedServerIds }, value => { values.push(value); return `$${values.length}` }) + const where = clauses.length ? ` WHERE ${clauses.join(" AND ")}` : "" + const res = await client.query<{ id: string }>(`SELECT COALESCE(max(id),0)::text AS id FROM router_log_entries${where}`, values) return res.rows[0].id } export async function readLogPage(filters: LogFilters & { allowedServerIds?: number[] }, options: { cursor?: string; watermark?: string; limit: number }) { const client = await pool.connect() try { await client.query("BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY") - const watermark = options.watermark ?? await logWatermark(client, filters.serverId) + const watermark = options.watermark ?? await logWatermark(client, filters.serverId, filters.allowedServerIds) const conditions = ["id <= $1::bigint"], values: unknown[] = [watermark] if (options.cursor) { const [at, id] = decodeLogCursor(options.cursor); conditions.push("(effective_at,id) < ($2::timestamptz,$3::bigint)"); values.push(at, id) } const where = logWhere(filters, conditions, values) @@ -48,11 +51,12 @@ export async function readLogSummary(filters: LogFilters & { allowedServerIds?: } return { total: Object.values(levels).reduce((a, b) => a + b, 0), levels, buckets: [...buckets.values()], settings: await getLogSettings(), states: await getLogStates(filters.serverId, filters.allowedServerIds) } } -export async function readLogEntry(id: string) { - return (await pool.query(`SELECT ${logColumns} FROM router_log_entries WHERE id=$1`, [id])).rows[0] ?? null +export async function readLogEntry(id: string, serverId?: number) { + const scoped = serverId !== undefined + return (await pool.query(`SELECT ${logColumns} FROM router_log_entries WHERE id=$1${scoped ? " AND server_id=$2" : ""}`, scoped ? [id, serverId] : [id])).rows[0] ?? null } -export async function readLogContext(id: string) { - const entry = await readLogEntry(id) +export async function readLogContext(id: string, serverId?: number) { + const entry = await readLogEntry(id, serverId) if (!entry) return null const before = (await pool.query(`SELECT ${logColumns} FROM router_log_entries WHERE server_id=$1 AND (effective_at,id)<($2::timestamptz,$3::bigint) ORDER BY effective_at DESC,router_log_entries.id DESC LIMIT 10`, [entry.serverId, entry.effectiveAt, id])).rows const after = (await pool.query(`SELECT ${logColumns} FROM router_log_entries WHERE server_id=$1 AND (effective_at,id)>($2::timestamptz,$3::bigint) ORDER BY effective_at,router_log_entries.id LIMIT 10`, [entry.serverId, entry.effectiveAt, id])).rows diff --git a/backend/src/services/router-logs-sql.test.ts b/backend/src/services/router-logs-sql.test.ts index 90a7798c..a294669c 100644 --- a/backend/src/services/router-logs-sql.test.ts +++ b/backend/src/services/router-logs-sql.test.ts @@ -8,7 +8,7 @@ import { logEntrySchema, logFiltersSchema, logPageSchema, logSummarySchema } fro import { pool } from "../db/index.js" import { env } from "../config.js" import { collectRouterLogsOnce, getLogStates, subscribeLogDevice, updateLogSettings } from "./router-logs-collector.js" -import { readLogContext, readLogPage, readLogStream, readLogSummary } from "./router-logs-query.js" +import { readLogContext, readLogEntry, readLogPage, readLogStream, readLogSummary } from "./router-logs-query.js" const schema = "test_router_logs_" + randomBytes(6).toString("hex") const admin = new pg.Pool({ connectionString: env.DATABASE_URL, connectionTimeoutMillis: 3000 }) @@ -50,6 +50,25 @@ try { const older = await readLogPage(filter, { limit: 200, cursor: page.nextCursor!, watermark: page.watermark }) assert.equal(older.entries.length, 101) assert.equal(new Set([...older.entries, ...page.entries].map(e => e.id)).size, 301) + const allServers = logFiltersSchema.parse({}) + const scoped = { ...allServers, allowedServerIds: [serverId] } + const scopedFirst = await readLogPage(scoped, { limit: 200 }) + const hiddenServerId = serverId + 1 + const { rows: [hidden] } = await pool.query<{ id: string }>(`INSERT INTO router_log_entries + (server_id,server_name,host,router_id,raw_time,received_at,effective_at,topics,level,message,generation) + VALUES($1,'hidden','192.0.2.2','*1','12:00:00',now(),now(),ARRAY['system'],'info','hidden event',0) RETURNING id::text`, [hiddenServerId]) + assert.equal((await readLogPage(allServers, { limit: 200 })).watermark, hidden.id, "Unscoped history keeps its global watermark") + assert.deepEqual(await readLogPage(scoped, { limit: 200 }), scopedFirst, "Hidden writes cannot advance a scoped watermark") + assert.deepEqual(await readLogPage({ ...scoped, serverId }, { limit: 200 }), scopedFirst) + const emptyPage = { entries: [], nextCursor: null, watermark: "0" } + assert.deepEqual(await readLogPage({ ...scoped, serverId: hiddenServerId }, { limit: 200 }), emptyPage) + assert.deepEqual(await readLogPage({ ...scoped, allowedServerIds: [] }, { limit: 200 }), emptyPage) + assert.deepEqual(await readLogPage(scoped, { limit: 200, cursor: scopedFirst.nextCursor!, watermark: scopedFirst.watermark }), older) + assert.equal((await readLogEntry(hidden.id))?.serverId, hiddenServerId) + assert.equal(await readLogEntry(hidden.id, serverId), null) + assert.equal(await readLogContext(hidden.id, serverId), null) + assert.equal((await readLogContext(hidden.id, hiddenServerId))?.entry.serverId, hiddenServerId) + await pool.query("DELETE FROM router_log_entries WHERE id=$1", [hidden.id]) await pool.query("UPDATE router_log_collection_state SET last_collected_at=now()-interval '1 minute'") await collectRouterLogsOnce() assert.equal((await readLogSummary(filter)).total, 301, "Repeated snapshot is not reimported") diff --git a/components/logs/logs-workspace.tsx b/components/logs/logs-workspace.tsx index bd617d43..7ca66a7f 100644 --- a/components/logs/logs-workspace.tsx +++ b/components/logs/logs-workspace.tsx @@ -128,6 +128,9 @@ function LogExplorer({ base, raw, selectedName, liveAvailable }: { base: string; {range === "custom" &&
{["from", "to"].map(key => )}
}
{collectorEnabled === undefined ? "Проверяем сборщик" : !collectorEnabled ? "Сборщик выключен" : result.streamError ? "Переподключение" : state?.lastError ? "Ошибка сбора" : live && result.stream ? "Live подключён" : "История"}Последний сбор: {formatLogTime(state?.lastCollectedAt ?? result.summary?.states.map(s => s.lastCollectedAt).filter((s): s is string => !!s).sort().at(-1) ?? null)}{live && }
+ {live && collectorEnabled === true && result.streamReceivedAt !== null &&

+ Получено (live-поток): +

} {collectorEnabled === false && История доступна, сбор новых событий выключенАдминистратор может включить коллектор в разделе «Сбор данных».} {(result.streamError || state?.lastError) && Live требует внимания{state?.lastError ?? result.streamError}} {state?.gapAt && Возможен разрыв историиБуфер изменился без пересечения с предыдущим снимком: {formatLogTime(state.gapAt)}. Количество недоступных событий неизвестно.} diff --git a/hooks/use-router-logs.ts b/hooks/use-router-logs.ts index 2eebbc22..93909da9 100644 --- a/hooks/use-router-logs.ts +++ b/hooks/use-router-logs.ts @@ -10,7 +10,8 @@ export function mergeLogEntries(previous: LogEntry[], incoming: LogEntry[], limi const message = (error: unknown) => error instanceof Error ? error.message : String(error) export function useRouterLogs(base: string, query: string, enabled: boolean, live: boolean, paused: boolean, windowMs = 0) { const [entries, setEntries] = useState([]), [page, setPage] = useState(null) - const [summary, setSummary] = useState(null), [stream, setStream] = useState(null) + const [summary, setSummary] = useState(null) + const [stream, setStream] = useState<{ batch: LogStream; receivedAt: number } | null>(null) const [loading, setLoading] = useState(enabled), [fetchingMore, setFetchingMore] = useState(false) const [error, setError] = useState(null), [summaryError, setSummaryError] = useState(null), [streamError, setStreamError] = useState(null) const [revision, setRevision] = useState(0), [pendingCount, setPendingCount] = useState(0) @@ -79,7 +80,7 @@ export function useRouterLogs(base: string, query: string, enabled: boolean, liv if (batch.entries.some(row => String(row.serverId) !== expectedServer) || batch.state && String(batch.state.serverId) !== expectedServer || BigInt(batch.cursor) < BigInt(cursor)) throw new Error("Live-поток относится к другому устройству или содержит некорректный курсор") heartbeat(); retry = 1000; cursor = batch.cursor if (ac.signal.aborted) return - setStream(batch); setStreamError(null) + setStream({ batch, receivedAt: Date.now() }); setStreamError(null) if (batch.entries.length) { if (pausedRef.current) { pending.current = mergeLogEntries(pending.current, batch.entries) @@ -119,5 +120,5 @@ export function useRouterLogs(base: string, query: string, enabled: boolean, liv finally { moreBusy.current = false; if (!signal?.aborted) setFetchingMore(false) } }, [base, query, enabled, page]) const reload = useCallback(() => { pending.current = []; setPendingCount(0); setRevision(n => n + 1) }, []) - return { entries, summary, stream, loading, fetchingMore, error, summaryError, streamError, pendingCount, hasMore: !!page?.nextCursor && entries.length < 5000, loadMore, reload, flush } + return { entries, summary, stream: stream?.batch ?? null, streamReceivedAt: stream?.receivedAt ?? null, loading, fetchingMore, error, summaryError, streamError, pendingCount, hasMore: !!page?.nextCursor && entries.length < 5000, loadMore, reload, flush } } diff --git a/tests/e2e/logs-context.spec.ts b/tests/e2e/logs-context.spec.ts index f2c52b40..10694e77 100644 --- a/tests/e2e/logs-context.spec.ts +++ b/tests/e2e/logs-context.spec.ts @@ -76,9 +76,13 @@ test("Логи: пауза буферизует новые события", asyn }) await page.goto("/logs?server=1") await expect.poll(() => started).toBe(true) + const receipt = page.getByLabel("Получение live-логов", { exact: true }).locator("time") + await expect(receipt).toHaveCount(0) await page.getByRole("button", { name: "Пауза", exact: true }).click() + const releasedAt = Date.now() release() await expect(page.getByRole("button", { name: "Новых событий: 1", exact: true })).toBeVisible() + await expect.poll(async () => Date.parse(await receipt.getAttribute("datetime", { timeout: 1000 }) ?? "")).toBeGreaterThanOrEqual(releasedAt) await expect(page.getByRole("button", { name: "Открыть событие 999", exact: true })).toHaveCount(0) await page.getByRole("button", { name: "Новых событий: 1", exact: true }).click() await expect(page.getByRole("button", { name: "Открыть событие 999", exact: true })).toBeVisible()