fix(geoip): проверить локальный кеш и ограничить загрузку баз

This commit is contained in:
Denozordec committed 2026-10-14 15:04:20 +07:00
1 parent 3c3d152c2d
commit 48cc65ec2b
16 files changed
+492 -31

No files matched your search

@@ -0,0 +1,27 @@
import assert from "node:assert/strict"
import { createServer } from "node:http"
import { downloadGeoipDatabase } from "./geoip-download.js"
let stalledClosed = false
const server = createServer((req, res) => {
if (req.url === "/slow") { res.writeHead(200); res.write("partial"); res.on("close", () => { stalledClosed = true }) }
else if (req.url === "/chunked") { res.writeHead(200); res.write("1234"); res.end("5678") }
else if (req.url === "/oversized") { res.writeHead(200, { "content-length": "9999" }); res.end("oversized") }
else if (req.url === "/304") { res.writeHead(304); res.end() }
else { res.writeHead(200); res.end("valid fixture") }
})
await new Promise<void>(resolve => server.listen(0, "127.0.0.1", resolve))
const address = server.address(); assert(address && typeof address !== "string")
const base = `http://127.0.0.1:${address.port}`
try {
const valid = await downloadGeoipDatabase(base, {}, fetch, 2000, 100)
assert.equal(valid.body.toString(), "valid fixture")
assert.equal((await downloadGeoipDatabase(base + "/304", {}, fetch, 2000, 100)).response.status, 304)
await assert.rejects(downloadGeoipDatabase(base + "/slow", {}, fetch, 120, 100), error => (error as Error).name === "AbortError")
await assert.rejects(downloadGeoipDatabase(base + "/oversized", {}, fetch, 2000, 5), /превышает/)
await assert.rejects(downloadGeoipDatabase(base + "/chunked", {}, fetch, 2000, 5), /превышает/)
// A failed request leaves no persistent timer or lock preventing the next successful download.
assert.equal((await downloadGeoipDatabase(base, {}, fetch, 2000, 100)).body.toString(), "valid fixture")
await new Promise(resolve => setTimeout(resolve, 20)); assert(stalledClosed)
console.log("GeoIP HTTP: whole-body timeout, declared/chunked limits,304 and recovery passed")
} finally { server.closeAllConnections(); await new Promise<void>(resolve => server.close(() => resolve())) }
+38
View File
@@ -0,0 +1,38 @@
const DOWNLOAD_TIMEOUT_MS = 120_000
const MAX_DOWNLOAD_BYTES = 64 * 1024 * 1024
/** A single deadline covers headers and streaming; a mirror cannot allocate an unbounded body. */
export async function downloadGeoipDatabase(
url: string, headers: Record<string, string>, fetcher: typeof fetch,
timeoutMs = DOWNLOAD_TIMEOUT_MS, maxBytes = MAX_DOWNLOAD_BYTES,
): Promise<{ response: Response; body: Buffer }> {
const controller = new AbortController()
const timer = setTimeout(() => controller.abort(), timeoutMs)
try {
const response = await fetcher(url, { headers, signal: controller.signal })
if (!response.ok || !response.body) {
await response.body?.cancel().catch(() => {})
return { response, body: Buffer.alloc(0) }
}
const size = Number(response.headers.get("content-length"))
if (Number.isFinite(size) && size > maxBytes) {
await response.body.cancel().catch(() => {})
throw new Error("Размер базы GeoIP превышает допустимый предел")
}
const reader = response.body.getReader(), chunks: Buffer[] = []
let received = 0
try {
for (;;) {
const { value, done } = await reader.read()
if (done) break
received += value.byteLength
if (received > maxBytes) {
await reader.cancel().catch(() => {})
throw new Error("Размер базы GeoIP превышает допустимый предел")
}
chunks.push(Buffer.from(value))
}
return { response, body: Buffer.concat(chunks, received) }
} finally { reader.releaseLock() }
} finally { clearTimeout(timer) }
}
+17 -18
View File
@@ -4,6 +4,7 @@ import { open, type AsnResponse, type CountryResponse } from "maxmind"
import type { GeoipUpdateRunSnapshot } from "../types/scheduler-run-snapshot.js"
import { SCHEDULER_RUN_SNAPSHOT_VERSION } from "../types/scheduler-run-snapshot.js"
import { getGeoipEtags, getGeoipSettings, touchGeoipRunMeta } from "./geoip-settings.js"
import { downloadGeoipDatabase } from "./geoip-download.js"
import {
GEOIP_ASN_FILE,
GEOIP_COUNTRY_FILE,
@@ -13,7 +14,6 @@ import {
/** Зеркало GeoLite2 без регистрации и ключей (см. README: GeoIP). */
const MIRROR_BASE = "https://github.com/P3TERX/GeoLite.mmdb/raw/download"
const DOWNLOAD_TIMEOUT_MS = 120_000
/** Пробный IP для валидации скачанной базы: Google DNS. */
const PROBE_IP = "8.8.8.8"
@@ -83,11 +83,7 @@ export async function collectGeoipUpdateOnce(
return emptySnapshot(sampledAt, true)
}
const settings = await getGeoipSettings()
if (!settings.enabled && !opts.force) {
return emptySnapshot(sampledAt, true)
}
// Acquire before the first await: concurrent requests must not both pass the gate.
updating = true
const snapshot: GeoipUpdateRunSnapshot = {
v: SCHEDULER_RUN_SNAPSHOT_VERSION,
@@ -101,6 +97,8 @@ export async function collectGeoipUpdateOnce(
}
try {
const settings = await getGeoipSettings()
if (!settings.enabled && !opts.force) return emptySnapshot(sampledAt, true)
const dir = geoipDir()
await mkdir(dir, { recursive: true })
const storedEtags = await getGeoipEtags()
@@ -114,25 +112,26 @@ export async function collectGeoipUpdateOnce(
const tmp = `${target}.tmp`
const prevEtag = storedEtags[file]
try {
const ac = new AbortController()
const timer = setTimeout(() => ac.abort(), DOWNLOAD_TIMEOUT_MS)
let res: Response
try {
res = await fetchImpl(`${MIRROR_BASE}/${file}`, {
headers: prevEtag ? { "If-None-Match": prevEtag } : {},
signal: ac.signal,
})
} finally {
clearTimeout(timer)
let cachedFileValid = false
if (prevEtag) {
try {
await (kind === "country" ? validateCountry(target) : validateAsn(target))
cachedFileValid = true
} catch { /* A missing or invalid local cache requires a full download. */ }
}
const { response: res, body } = await downloadGeoipDatabase(
`${MIRROR_BASE}/${file}`,
prevEtag && cachedFileValid ? { "If-None-Match": prevEtag } : {},
fetchImpl,
)
if (res.status === 304) {
if (!cachedFileValid || !prevEtag) throw new Error("Зеркало вернуло HTTP 304 без проверенной локальной базы")
snapshot.skippedUnchanged += 1
if (prevEtag) etags[file] = prevEtag
continue
}
if (!res.ok) throw new Error(`HTTP ${res.status}`)
const etag = res.headers.get("etag") ?? ""
const body = Buffer.from(await res.arrayBuffer())
snapshot.bytes += body.byteLength
await writeFile(tmp, body)
buildAt[kind] =
@@ -161,7 +160,7 @@ export async function collectGeoipUpdateOnce(
await touchGeoipRunMeta({
lastCheckAt: sampledAt,
lastSuccessAt: snapshot.errors.length ? null : sampledAt,
...(snapshot.errors.length ? {} : { lastSuccessAt: sampledAt }),
lastError: snapshot.errors.length ? snapshot.errors.join("; ") : null,
countryBuildAt: buildAt.country,
asnBuildAt: buildAt.asn,
@@ -141,7 +141,7 @@ const dims = minuteDimsSnapshotForTests()
assert.equal(dims.get("country")?.get("US")?.bytes, 1000)
assert.equal(dims.get("asn")?.get("15169")?.bytes, 1000)
// ── коллектор: 304 → обе базы без изменений ───────────────────────────────────
// ── коллектор: 304 без локального cache не является успехом ──────────────────
disableGeoipDbForTests()
resetGeoipUpdateForTests()
const geoipDir = mkdtempSync(path.join(tmpdir(), "mm-geoip-test-"))
@@ -149,7 +149,9 @@ process.env.GEOIP_DIR = geoipDir
setGeoipFetchForTests(async () => new Response(null, { status: 304 }))
let snap = await collectGeoipUpdateOnce({ force: true })
assert.equal(snap.skippedUnchanged, 2)
assert.equal(snap.skippedUnchanged, 0)
assert.equal(snap.errors.length, 2)
assert(snap.errors.every(message => message.includes("304 без проверенной локальной базы")))
assert.equal(snap.downloaded, 0)
assert.equal(existsSync(path.join(geoipDir, "GeoLite2-Country.mmdb")), false)
@@ -195,6 +197,17 @@ assert.equal(readFileSync(asnPath, "utf8"), "good-asn", "asn не тронут")
assert.equal(existsSync(`${countryPath}.tmp`), false)
assert.equal(existsSync(`${asnPath}.tmp`), false)
// Lock is acquired before settings lookup yields; a concurrent forced run rejects.
let releaseFetch!: () => void
const heldFetch = new Promise<void>(resolve => { releaseFetch = resolve })
setGeoipFetchForTests(async () => { await heldFetch; return new Response(null, { status: 503 }) })
const firstRun = collectGeoipUpdateOnce({ force: true })
await assert.rejects(collectGeoipUpdateOnce({ force: true }), error => (error as { statusCode?: number }).statusCode === 409)
assert.equal((await collectGeoipUpdateOnce()).skipped, true)
releaseFetch()
await firstRun
assert.equal((await collectGeoipUpdateOnce({ force: true })).errors.length, 2, "lock released after errors")
rmSync(geoipDir, { recursive: true, force: true })
delete process.env.GEOIP_DIR
resetGeoipUpdateForTests()