Harden server state and config persistence

2026-07-11 21:07:11 +03:00
parent e6b21ed8a9
commit c9223aa3a9
6 changed files with 338 additions and 50 deletions
--- a/docs/recovery/state-recovery.md
+++ 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 . <backup-file>`.
+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.
--- 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);
--- a/src/server/services/stateStore.js
+++ 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,
+  });
+}
--- 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() {
--- 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;
 }

--- a/test/server/state-store.test.js
+++ 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/);
+});