From c9223aa3a9da2d01e72bdc19eb96bfb6bb885ff1 Mon Sep 17 00:00:00 2001 From: Dmitriy Petrov Date: Sat, 11 Jul 2026 21:07:11 +0300 Subject: [PATCH] Harden server state and config persistence --- docs/recovery/state-recovery.md | 36 +++++++ src/server/index.js | 107 ++++++++++++--------- src/server/services/stateStore.js | 150 ++++++++++++++++++++++++++++++ src/server/singbox.js | 9 +- src/server/subscription.js | 3 +- test/server/state-store.test.js | 83 +++++++++++++++++ 6 files changed, 338 insertions(+), 50 deletions(-) create mode 100644 docs/recovery/state-recovery.md create mode 100644 src/server/services/stateStore.js create mode 100644 test/server/state-store.test.js diff --git a/docs/recovery/state-recovery.md b/docs/recovery/state-recovery.md new file mode 100644 index 0000000..6d2bdf7 --- /dev/null +++ b/docs/recovery/state-recovery.md @@ -0,0 +1,36 @@ +# Harbor state recovery + +Harbor keeps the existing data paths and volumes. `state.json` now uses `schemaVersion: 1`; subscription cache, generated sing-box config and HWID keep their existing filenames. + +## Atomic writes + +Persistent files are written to a unique temporary file in the same directory, flushed with `fsync`, closed and atomically renamed over the target. A failure before rename leaves the previous target untouched and removes the temporary file. + +## Migration + +On startup, a legacy `state.json` without `schemaVersion` is normalized and migrated to v1. Before replacement Harbor saves the original beside it: + +```text +state.json.backup-v0-2026-07-11T12-00-00-000Z +``` + +The v1 migration preserves existing fields, adds normalized revision, selection and server fields, and does not rename the volume. Older Harbor builds ignore the additional `schemaVersion` field, but the backup is the safest rollback source. + +## Corrupt JSON + +If `state.json` cannot be parsed, Harbor renames the exact damaged bytes to: + +```text +state.json.corrupt-2026-07-11T12-00-00-000Z +``` + +It then creates a valid empty v1 state and reports `storage-recovery` through `snapshot.operation`. A corrupt subscription cache is preserved with the same suffix and reported in control logs. + +Recovery should be performed while Harbor is stopped: + +1. Copy the whole data directory before changing anything. +2. Inspect a backup with `jq . `. +3. Restore only a valid JSON backup to the original filename. +4. Start Harbor and verify `GET /api/state` before applying or importing anything. + +Generated config rollback also uses the atomic writer. No automatic recovery tries to guess missing subscription credentials or repair semantically invalid sing-box configuration. diff --git a/src/server/index.js b/src/server/index.js index 8e50703..b63cc64 100644 --- a/src/server/index.js +++ b/src/server/index.js @@ -20,6 +20,7 @@ import { buildSharedProxyInfo } from './sharedProxy.js'; import { buildGatewayConfig, removeSingboxConfig, + restoreSingboxConfig, writeSingboxConfig, } from './singbox.js'; import { fetchSubscription, getHwid, selectRefreshedServer } from './subscription.js'; @@ -29,6 +30,7 @@ import { withStateV0Compatibility, } from '../shared/contracts/state.js'; import { HarborError, normalizeHarborError } from '../shared/errors.js'; +import { createJsonStore, createStateStore } from './services/stateStore.js'; const MAX_BODY_BYTES = 1_000_000; const SUBSCRIPTION_REFRESH_INTERVAL_MS = 15 * 60 * 1000; @@ -36,6 +38,30 @@ const GATEWAY_DISCOVERY_INTERVAL_MS = 5_000; fs.mkdirSync(settings.dataDir, { recursive: true }); +const stateStore = createStateStore(settings.statePath); +const subscriptionCacheStore = createJsonStore({ + filePath: settings.subscriptionCachePath, + defaultValue: null, +}); +let cacheRecoveryLogged = false; + +function readSubscriptionCache() { + const cached = subscriptionCacheStore.read(); + if (subscriptionCacheStore.recovery && !cacheRecoveryLogged) { + cacheRecoveryLogged = true; + console.warn(`[storage] corrupt subscription cache recovered; backup: ${subscriptionCacheStore.recovery.backupPath}`); + } + return cached; +} + +const initialStoredState = stateStore.read(); +if (stateStore.migration) { + console.log(`[storage] state migrated to v${stateStore.migration.toVersion}; backup: ${stateStore.migration.backupPath}`); +} +if (stateStore.recovery) { + console.warn(`[storage] corrupt state recovered; backup: ${stateStore.recovery.backupPath}`); +} + const remoteDataplane = settings.appMode === 'gateway' && Boolean(process.env.DATAPLANE_SOCKET); const singboxRuntime = remoteDataplane ? createDataplaneClient(settings.dataplaneSocket) @@ -50,33 +76,22 @@ let gatewayDiscoveryPromise = null; let gatewayDiscoveryTimer = null; let gatewayAutoState = createGatewayAutoState(); let controlOperation = Promise.resolve(); -let operationState = { kind: null, status: 'idle', startedAt: null, error: null }; -let revision = 0; - -function readJson(filePath, fallback) { - try { - return fs.existsSync(filePath) - ? JSON.parse(fs.readFileSync(filePath, 'utf8')) - : fallback; - } catch { - return fallback; - } -} - -function writeJson(filePath, value) { - fs.mkdirSync(path.dirname(filePath), { recursive: true }); - fs.writeFileSync(filePath, JSON.stringify(value, null, 2), 'utf8'); -} - -revision = normalizeStoredState(readJson(settings.statePath, {})).revision; +let operationState = stateStore.recovery ? { + kind: 'storage-recovery', + status: 'failed', + startedAt: stateStore.recovery.recoveredAt, + error: `Повреждённый state сохранён: ${path.basename(stateStore.recovery.backupPath)}`, +} : { kind: null, status: 'idle', startedAt: null, error: null }; +let revision = normalizeStoredState(initialStoredState).revision; function updateStoredState(update) { - const current = normalizeStoredState(readJson(settings.statePath, {})); - const next = normalizeStoredState(update(current)); - revision = Math.max(revision, current.revision) + 1; - next.revision = revision; - writeJson(settings.statePath, next); - return next; + return stateStore.update((stored) => { + const current = normalizeStoredState(stored); + const next = normalizeStoredState(update(current)); + revision = Math.max(revision, current.revision) + 1; + next.revision = revision; + return next; + }); } async function withOperation(kind, operation) { @@ -186,7 +201,7 @@ const startSingbox = () => singboxRuntime.apply(); async function publicState() { const runtime = await singboxRuntime.refresh(); - const state = normalizeStoredState(readJson(settings.statePath, {})); + const state = normalizeStoredState(stateStore.read()); const gatewayAutoEnabled = state.gatewayAutoEnabled !== false; const configExists = fs.existsSync(settings.configPath); const snapshot = createStateSnapshot({ @@ -208,8 +223,8 @@ async function publicState() { } function writeCurrentConfig() { - const state = readJson(settings.statePath, {}); - const cached = readJson(settings.subscriptionCachePath, null); + const state = stateStore.read(); + const cached = readSubscriptionCache(); if (!state.selectedTag || !cached?.config) return false; writeSingboxConfig(buildActiveConfig(cached.config, state.selectedTag)); return true; @@ -235,7 +250,7 @@ async function applyGatewayAutoState(nextState, { reconfigure = true } = {}) { } catch (error) { gatewayAutoState = previousState; if (previousConfig === null) removeSingboxConfig(); - else fs.writeFileSync(settings.configPath, previousConfig, 'utf8'); + else restoreSingboxConfig(previousConfig); throw error; } @@ -250,7 +265,7 @@ function refreshGatewayAutoMode({ reconfigure = true } = {}) { if (gatewayDiscoveryPromise) return gatewayDiscoveryPromise; gatewayDiscoveryPromise = serializeControl(async () => { - const state = readJson(settings.statePath, {}); + const state = stateStore.read(); const network = state.subscriptionUrl ? readHostNetworkState(settings.hostNetworkStatePath) : null; @@ -283,7 +298,7 @@ function refreshGatewayAutoMode({ reconfigure = true } = {}) { port: settings.gatewayPresencePort, subscriptionUrl: state.subscriptionUrl, }); - const latestState = readJson(settings.statePath, {}); + const latestState = stateStore.read(); const latestNetwork = latestState.subscriptionUrl ? readHostNetworkState(settings.hostNetworkStatePath) : null; @@ -303,7 +318,7 @@ function refreshGatewayAutoMode({ reconfigure = true } = {}) { ); } catch (error) { const reason = error?.message || 'Gateway presence check failed'; - const latestState = readJson(settings.statePath, {}); + const latestState = stateStore.read(); const latestNetwork = latestState.subscriptionUrl ? readHostNetworkState(settings.hostNetworkStatePath) : null; @@ -334,7 +349,7 @@ function refreshGatewayAutoMode({ reconfigure = true } = {}) { } async function applySelectedServer(selectedTag, { persist = true } = {}) { - const cached = readJson(settings.subscriptionCachePath, null); + const cached = readSubscriptionCache(); if (!cached?.config) throw new HarborError('CONFIG_INVALID'); const nextConfig = buildActiveConfig(cached.config, selectedTag); @@ -354,7 +369,7 @@ async function applySelectedServer(selectedTag, { persist = true } = {}) { await startSingbox(); } catch (error) { if (previousConfig === null) removeSingboxConfig(); - else fs.writeFileSync(settings.configPath, previousConfig, 'utf8'); + else restoreSingboxConfig(previousConfig); throw error; } if (persist) { @@ -370,7 +385,7 @@ function refreshSavedSubscription() { if (subscriptionRefreshPromise) return subscriptionRefreshPromise; subscriptionRefreshPromise = (async () => { - const initialState = readJson(settings.statePath, {}); + const initialState = stateStore.read(); if (!initialState.subscriptionUrl) { throw new HarborError('SUBSCRIPTION_INVALID'); } @@ -378,13 +393,13 @@ function refreshSavedSubscription() { const subscriptionUrl = initialState.subscriptionUrl; const parsed = await fetchSubscription(subscriptionUrl); return serializeControl(async () => { - const currentState = readJson(settings.statePath, {}); + const currentState = stateStore.read(); if (currentState.subscriptionUrl !== subscriptionUrl) { throw new HarborError('STATE_CONFLICT'); } const selectedTag = selectRefreshedServer(currentState.selectedTag, parsed.servers); - const previousCache = readJson(settings.subscriptionCachePath, null); + const previousCache = readSubscriptionCache(); const previousConfig = fs.existsSync(settings.configPath) ? fs.readFileSync(settings.configPath, 'utf8') : null; @@ -394,7 +409,7 @@ function refreshSavedSubscription() { buildActiveConfig(parsed.config, selectedTag), ) ); - writeJson(settings.subscriptionCachePath, { url: subscriptionUrl, ...parsed }); + subscriptionCacheStore.write({ url: subscriptionUrl, ...parsed }); try { if (singboxRuntime.running && activeConfigChanged) { @@ -406,10 +421,10 @@ function refreshSavedSubscription() { removeSingboxConfig(); } } catch (error) { - if (previousCache) writeJson(settings.subscriptionCachePath, previousCache); - else fs.rmSync(settings.subscriptionCachePath, { force: true }); + if (previousCache) subscriptionCacheStore.write(previousCache); + else subscriptionCacheStore.remove(); if (previousConfig === null) removeSingboxConfig(); - else fs.writeFileSync(settings.configPath, previousConfig, 'utf8'); + else restoreSingboxConfig(previousConfig); throw error; } @@ -459,7 +474,7 @@ async function handleApi(req, res) { const requestUrl = new URL(req.url, `http://localhost:${settings.port}`); if (req.method === 'GET' && requestUrl.pathname === '/api/gateway-presence') { - const state = readJson(settings.statePath, {}); + const state = stateStore.read(); return sendJson(res, 200, buildGatewayPresence({ appMode: settings.appMode, subscriptionUrl: state.subscriptionUrl, @@ -469,7 +484,7 @@ async function handleApi(req, res) { } if (req.method === 'POST' && req.url === '/api/servers/ping-all') { - const state = readJson(settings.statePath, {}); + const state = stateStore.read(); const results = await Promise.all((state.servers || []).map(async (server) => ({ tag: String(server.tag || '').trim(), ...await tcpPing(server.server, server.server_port), @@ -486,7 +501,7 @@ async function handleApi(req, res) { await serializeControl(async () => { await stopSingbox(); removeSingboxConfig(); - writeJson(settings.subscriptionCachePath, { url: normalizedUrl, ...result }); + subscriptionCacheStore.write({ url: normalizedUrl, ...result }); updateStoredState((state) => ({ subscriptionUrl: normalizedUrl, gatewayAutoEnabled: state.gatewayAutoEnabled !== false, @@ -541,7 +556,7 @@ async function handleApi(req, res) { await withOperation('subscription-forget', () => serializeControl(async () => { await stopSingbox(); removeSingboxConfig(); - fs.rmSync(settings.subscriptionCachePath, { force: true }); + subscriptionCacheStore.remove(); updateStoredState(() => ({})); gatewayAutoState = createGatewayAutoState(); })); @@ -638,7 +653,7 @@ server.listen(settings.port, '0.0.0.0', () => { }); subscriptionRefreshTimer = setInterval(() => { - if (!readJson(settings.statePath, {}).subscriptionUrl) return; + if (!stateStore.read().subscriptionUrl) return; refreshSavedSubscription() .catch((error) => console.warn(`[control] подписка не обновлена: ${error.message}`)); }, SUBSCRIPTION_REFRESH_INTERVAL_MS); diff --git a/src/server/services/stateStore.js b/src/server/services/stateStore.js new file mode 100644 index 0000000..e29b635 --- /dev/null +++ b/src/server/services/stateStore.js @@ -0,0 +1,150 @@ +import crypto from 'node:crypto'; +import fs from 'node:fs'; +import path from 'node:path'; +import { normalizeStoredState } from '../../shared/contracts/state.js'; + +export const STATE_SCHEMA_VERSION = 1; + +const clone = (value) => structuredClone(value); +const stamp = (value) => value.toISOString().replace(/[:.]/g, '-'); + +function syncDirectory(directory) { + let descriptor; + try { + descriptor = fs.openSync(directory, 'r'); + fs.fsyncSync(descriptor); + } catch (error) { + if (!['EINVAL', 'ENOTSUP', 'EPERM'].includes(error.code)) throw error; + } finally { + if (descriptor !== undefined) fs.closeSync(descriptor); + } +} + +export function atomicWriteFile(filePath, contents, { beforeRename, mode } = {}) { + const directory = path.dirname(filePath); + fs.mkdirSync(directory, { recursive: true }); + const temporaryPath = path.join( + directory, + `.${path.basename(filePath)}.${process.pid}.${crypto.randomUUID()}.tmp`, + ); + const fileMode = mode ?? (fs.existsSync(filePath) ? fs.statSync(filePath).mode & 0o777 : 0o666); + let descriptor; + + try { + descriptor = fs.openSync(temporaryPath, 'wx', fileMode); + fs.writeFileSync(descriptor, contents, 'utf8'); + fs.fsyncSync(descriptor); + fs.closeSync(descriptor); + descriptor = undefined; + beforeRename?.(temporaryPath, filePath); + fs.renameSync(temporaryPath, filePath); + syncDirectory(directory); + } finally { + if (descriptor !== undefined) fs.closeSync(descriptor); + fs.rmSync(temporaryPath, { force: true }); + } +} + +export function atomicWriteJson(filePath, value, options) { + atomicWriteFile(filePath, JSON.stringify(value, null, 2), options); +} + +export function migrateStoredState(value) { + const stored = value && typeof value === 'object' && !Array.isArray(value) ? value : {}; + const version = Number.isSafeInteger(stored.schemaVersion) ? stored.schemaVersion : 0; + if (version < 0 || version > STATE_SCHEMA_VERSION) { + throw new Error(`Unsupported Harbor state schemaVersion: ${version}`); + } + return { + ...normalizeStoredState(stored), + schemaVersion: STATE_SCHEMA_VERSION, + }; +} + +export function createJsonStore({ + filePath, + defaultValue, + migrate = (value) => value, + initializeMissing = false, + backupWhen = () => false, + now = () => new Date(), +} = {}) { + let recovery = null; + let migration = null; + + function write(value, options) { + const migrated = migrate(clone(value)); + atomicWriteJson(filePath, migrated, options); + return clone(migrated); + } + + function read() { + if (!fs.existsSync(filePath)) { + const initial = migrate(clone(defaultValue)); + return initializeMissing ? write(initial) : clone(initial); + } + + const raw = fs.readFileSync(filePath, 'utf8'); + let parsed; + try { + parsed = JSON.parse(raw); + } catch (cause) { + const backupPath = `${filePath}.corrupt-${stamp(now())}`; + fs.renameSync(filePath, backupPath); + try { + const recovered = write(defaultValue); + recovery = { kind: 'corrupt-json', backupPath, recoveredAt: now().toISOString() }; + return recovered; + } catch (error) { + fs.renameSync(backupPath, filePath); + throw new AggregateError([cause, error], `Failed to recover corrupt JSON: ${filePath}`); + } + } + + const migrated = migrate(clone(parsed)); + if (JSON.stringify(migrated) !== JSON.stringify(parsed)) { + if (backupWhen(parsed, migrated)) { + const fromVersion = Number.isSafeInteger(parsed?.schemaVersion) ? parsed.schemaVersion : 0; + const backupPath = `${filePath}.backup-v${fromVersion}-${stamp(now())}`; + atomicWriteFile(backupPath, raw, { mode: fs.statSync(filePath).mode & 0o777 }); + migration = { + fromVersion, + toVersion: migrated.schemaVersion, + backupPath, + migratedAt: now().toISOString(), + }; + } + atomicWriteJson(filePath, migrated); + } + return clone(migrated); + } + + function update(mutator) { + // ponytail: sync mutators serialize in Node's event loop; add a queue only if updates must await I/O. + const next = mutator(read()); + if (next && typeof next.then === 'function') { + throw new TypeError('State store mutator must be synchronous'); + } + return write(next); + } + + return { + read, + write, + update, + remove: () => fs.rmSync(filePath, { force: true }), + get recovery() { return recovery; }, + get migration() { return migration; }, + }; +} + +export function createStateStore(filePath, options = {}) { + return createJsonStore({ + filePath, + defaultValue: {}, + migrate: migrateStoredState, + initializeMissing: true, + backupWhen: (before, after) => before?.schemaVersion !== after.schemaVersion, + ...options, + }); +} diff --git a/src/server/singbox.js b/src/server/singbox.js index 2cbe2be..9ef1360 100644 --- a/src/server/singbox.js +++ b/src/server/singbox.js @@ -1,7 +1,7 @@ import fs from 'node:fs'; -import path from 'node:path'; import { settings } from './config.js'; import { HarborError } from '../shared/errors.js'; +import { atomicWriteFile, atomicWriteJson } from './services/stateStore.js'; const PROXY_TYPES = new Set(['vless', 'vmess', 'trojan', 'shadowsocks', 'hysteria2']); const MIXED_INBOUND = 'mixed-in'; @@ -77,8 +77,11 @@ export function buildGatewayConfig(subscriptionConfig, selectedTag, { clientDire } export function writeSingboxConfig(config) { - fs.mkdirSync(path.dirname(settings.configPath), { recursive: true }); - fs.writeFileSync(settings.configPath, JSON.stringify(config, null, 2), 'utf8'); + atomicWriteJson(settings.configPath, config); +} + +export function restoreSingboxConfig(contents) { + atomicWriteFile(settings.configPath, contents); } export function removeSingboxConfig() { diff --git a/src/server/subscription.js b/src/server/subscription.js index 8453f66..ebf30f2 100644 --- a/src/server/subscription.js +++ b/src/server/subscription.js @@ -2,6 +2,7 @@ import crypto from 'node:crypto'; import fs from 'node:fs'; import { settings } from './config.js'; import { HarborError } from '../shared/errors.js'; +import { atomicWriteFile } from './services/stateStore.js'; const PROXY_TYPES = new Set(['vless', 'vmess', 'trojan', 'shadowsocks', 'hysteria2']); @@ -11,7 +12,7 @@ export function getHwid() { return fs.readFileSync(settings.hwidPath, 'utf8').trim(); } const hwid = crypto.randomBytes(8).toString('hex'); - fs.writeFileSync(settings.hwidPath, hwid, 'utf8'); + atomicWriteFile(settings.hwidPath, hwid); return hwid; } diff --git a/test/server/state-store.test.js b/test/server/state-store.test.js new file mode 100644 index 0000000..b4be037 --- /dev/null +++ b/test/server/state-store.test.js @@ -0,0 +1,83 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; + +import { + atomicWriteJson, + createStateStore, + STATE_SCHEMA_VERSION, +} from '../../src/server/services/stateStore.js'; + +const fixture = (t) => { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'harbor-state-store-')); + t.after(() => fs.rmSync(directory, { recursive: true, force: true })); + return path.join(directory, 'state.json'); +}; + +test('a failure before rename preserves the last successful file', (t) => { + const filePath = fixture(t); + atomicWriteJson(filePath, { revision: 1 }); + + assert.throws( + () => atomicWriteJson(filePath, { revision: 2 }, { + beforeRename: () => { throw new Error('injected failure'); }, + }), + /injected failure/, + ); + assert.deepEqual(JSON.parse(fs.readFileSync(filePath, 'utf8')), { revision: 1 }); + assert.equal( + fs.readdirSync(path.dirname(filePath)).some((name) => name.endsWith('.tmp')), + false, + ); +}); + +test('legacy state migrates to schema v1 and keeps a backup', (t) => { + const filePath = fixture(t); + const legacy = { + revision: 7, + selectedTag: 'nl', + servers: [{ tag: 'nl' }], + }; + fs.writeFileSync(filePath, JSON.stringify(legacy)); + + const store = createStateStore(filePath, { + now: () => new Date('2026-07-11T12:00:00.000Z'), + }); + const migrated = store.read(); + + assert.equal(migrated.schemaVersion, STATE_SCHEMA_VERSION); + assert.equal(migrated.appliedTag, 'nl'); + assert.equal(store.migration.fromVersion, 0); + assert.deepEqual(JSON.parse(fs.readFileSync(store.migration.backupPath, 'utf8')), legacy); + assert.equal(JSON.parse(fs.readFileSync(filePath, 'utf8')).schemaVersion, 1); +}); + +test('corrupt JSON is preserved and replaced with an explicit recovery state', (t) => { + const filePath = fixture(t); + fs.writeFileSync(filePath, '{broken'); + + const store = createStateStore(filePath, { + now: () => new Date('2026-07-11T12:00:00.000Z'), + }); + const recovered = store.read(); + + assert.equal(recovered.schemaVersion, 1); + assert.equal(recovered.revision, 0); + assert.equal(store.recovery.kind, 'corrupt-json'); + assert.equal(fs.readFileSync(store.recovery.backupPath, 'utf8'), '{broken'); + assert.equal(JSON.parse(fs.readFileSync(filePath, 'utf8')).schemaVersion, 1); +}); + +test('concurrent updates are serialized without lost values', async (t) => { + const store = createStateStore(fixture(t)); + store.read(); + + await Promise.all(Array.from({ length: 50 }, () => Promise.resolve().then(() => ( + store.update((state) => ({ ...state, counter: (state.counter || 0) + 1 })) + )))); + + assert.equal(store.read().counter, 50); + assert.throws(() => store.update(async (state) => state), /must be synchronous/); +});