fix: preserve LMDB slashing protection (#23145)

- Preserve local validator slashing-protection records across the known
LMDB schema 1 -> 2 migration.
- Add a fail-closed schema mismatch policy for versioned stores and wire
it into signing protection.
- Add regression coverage for preserving legacy duty records and
refusing newer stored schemas.

Fixes
[A-1029](https://linear.app/aztec-labs/issue/A-1029/prevent-lmdb-slashing-protection-reset-on-schema-mismatch)
Fixes AztecProtocol/aztec-claude#888

Author: spypsy

yarn-project/kv-store/src/lmdb-v2/factory.ts

@@ -1,6 +1,6 @@
 import { EthAddress } from '@aztec/foundation/eth-address';
 import { type LoggerBindings, createLogger } from '@aztec/foundation/log';
-import { DatabaseVersionManager } from '@aztec/stdlib/database-version/manager';
+import { DatabaseVersionManager, type SchemaVersionMismatchPolicy } from '@aztec/stdlib/database-version/manager';
 import type { DataStoreConfig } from '@aztec/stdlib/kv-store';
 
 import { mkdir, mkdtemp, rm } from 'fs/promises';
@@ -11,11 +11,18 @@ import { AztecLMDBStoreV2 } from './store.js';
 
 const MAX_READERS = 16;
 
+/** Optional versioning hooks for persistent LMDB stores. */
+export type CreateStoreOptions = {
+  onUpgrade?: (dataDir: string, currentVersion: number, latestVersion: number) => Promise<void>;
+  schemaVersionMismatchPolicy?: SchemaVersionMismatchPolicy;
+};
+
 export async function createStore(
   name: string,
   schemaVersion: number,
   config: DataStoreConfig,
   bindings?: LoggerBindings,
+  options: CreateStoreOptions = {},
 ): Promise<AztecLMDBStoreV2> {
   const log = createLogger('kv-store:lmdb-v2:' + name, bindings);
   const { dataDirectory, l1Contracts } = config;
@@ -35,6 +42,8 @@ export async function createStore(
       dataDirectory: subDir,
       onOpen: dbDirectory =>
         AztecLMDBStoreV2.new(dbDirectory, config.dataStoreMapSizeKb, MAX_READERS, () => Promise.resolve(), bindings),
+      onUpgrade: options.onUpgrade,
+      schemaVersionMismatchPolicy: options.schemaVersionMismatchPolicy,
     });
 
     log.info(

yarn-project/stdlib/src/database-version/version_manager.test.ts

@@ -15,6 +15,7 @@ describe('VersionManager', () => {
   let versionManager: DatabaseVersionManager<object>;
   let currentVersion: number;
   let rollupAddress: EthAddress;
+  let expectVersionFileWritten: boolean;
 
   beforeEach(() => {
     fs = {
@@ -28,6 +29,7 @@ describe('VersionManager', () => {
 
     currentVersion = 42;
     rollupAddress = EthAddress.random();
+    expectVersionFileWritten = true;
 
     openSpy = jest.fn(() => Promise.resolve({}));
     upgradeSpy = jest.fn(() => Promise.resolve());
@@ -43,6 +45,9 @@ describe('VersionManager', () => {
 
   describe('open', () => {
     afterEach(() => {
+      if (!expectVersionFileWritten) {
+        return;
+      }
       // Verify version file was created
       expect(fs.writeFile).toHaveBeenCalledWith(
         join(tempDir, DatabaseVersionManager.VERSION_FILE),
@@ -96,6 +101,26 @@ describe('VersionManager', () => {
         expect(upgradeSpy).not.toHaveBeenCalled();
       });
 
+      it('unless schema mismatches are configured to throw', async () => {
+        fs.readFile.mockResolvedValueOnce(new DatabaseVersion(currentVersion + 1, rollupAddress).toBuffer());
+        versionManager = new DatabaseVersionManager({
+          schemaVersion: currentVersion,
+          rollupAddress,
+          dataDirectory: tempDir,
+          onOpen: openSpy,
+          onUpgrade: upgradeSpy,
+          schemaVersionMismatchPolicy: 'throw',
+          fileSystem: fs,
+        });
+        expectVersionFileWritten = false;
+
+        await expect(versionManager.open()).rejects.toThrow(
+          `stored schema version ${currentVersion + 1} is incompatible with expected schema version ${currentVersion}`,
+        );
+        expect(fs.rm).not.toHaveBeenCalled();
+        expect(openSpy).not.toHaveBeenCalled();
+      });
+
       it('when the upgrade fails', async () => {
         fs.readFile.mockResolvedValueOnce(new DatabaseVersion(currentVersion - 1, rollupAddress).toBuffer());
         upgradeSpy.mockRejectedValueOnce(new Error('Test: failed upgrade'));

yarn-project/stdlib/src/database-version/version_manager.ts

@@ -9,13 +9,15 @@ import { DatabaseVersion } from './database_version.js';
 export type DatabaseVersionManagerFs = Pick<typeof fs, 'readFile' | 'writeFile' | 'rm' | 'mkdir'>;
 
 export const DATABASE_VERSION_FILE_NAME = 'db_version';
+export type SchemaVersionMismatchPolicy = 'reset' | 'throw';
 
 export type DatabaseVersionManagerOptions<T> = {
   schemaVersion: number;
   rollupAddress: EthAddress;
   dataDirectory: string;
   onOpen: (dataDir: string) => Promise<T>;
   onUpgrade?: (dataDir: string, currentVersion: number, latestVersion: number) => Promise<void>;
+  schemaVersionMismatchPolicy?: SchemaVersionMismatchPolicy;
   fileSystem?: DatabaseVersionManagerFs;
   log?: Logger;
 };
@@ -34,6 +36,7 @@ export class DatabaseVersionManager<T> {
   private dataDirectory: string;
   private onOpen: (dataDir: string) => Promise<T>;
   private onUpgrade?: (dataDir: string, currentVersion: number, latestVersion: number) => Promise<void>;
+  private schemaVersionMismatchPolicy: SchemaVersionMismatchPolicy;
   private fileSystem: DatabaseVersionManagerFs;
   private log: Logger;
 
@@ -45,6 +48,7 @@ export class DatabaseVersionManager<T> {
    * @param dataDirectory - The directory where version information will be stored
    * @param onOpen - A callback to the open the database at the given location
    * @param onUpgrade - An optional callback to upgrade the database before opening. If not provided it will reset the database
+   * @param schemaVersionMismatchPolicy - Whether schema mismatches should reset data or throw
    * @param fileSystem - An interface to access the filesystem
    * @param log - Optional custom logger
    * @param options - Configuration options
@@ -55,6 +59,7 @@ export class DatabaseVersionManager<T> {
     dataDirectory,
     onOpen,
     onUpgrade,
+    schemaVersionMismatchPolicy = 'reset',
     fileSystem = fs,
     log = createLogger(`foundation:version-manager`),
   }: DatabaseVersionManagerOptions<T>) {
@@ -68,6 +73,7 @@ export class DatabaseVersionManager<T> {
     this.dataDirectory = dataDirectory;
     this.onOpen = onOpen;
     this.onUpgrade = onUpgrade;
+    this.schemaVersionMismatchPolicy = schemaVersionMismatchPolicy;
     this.fileSystem = fileSystem;
     this.log = log;
   }
@@ -115,10 +121,21 @@ export class DatabaseVersionManager<T> {
         try {
           await this.onUpgrade(this.dataDirectory, storedVersion.schemaVersion, this.currentVersion.schemaVersion);
         } catch (error) {
+          if (this.schemaVersionMismatchPolicy === 'throw') {
+            throw new Error(
+              `Failed to upgrade database at ${this.dataDirectory} from schema version ${storedVersion.schemaVersion} to ${this.currentVersion.schemaVersion}`,
+              { cause: error },
+            );
+          }
           this.log.error(`Failed to upgrade: ${error}. Falling back to reset.`);
           needsReset = true;
         }
       } else if (cmp !== 0) {
+        if (this.schemaVersionMismatchPolicy === 'throw') {
+          throw new Error(
+            `Cannot open database at ${this.dataDirectory}: stored schema version ${storedVersion.schemaVersion} is incompatible with expected schema version ${this.currentVersion.schemaVersion}`,
+          );
+        }
         if (shouldLogDataReset) {
           this.log.info(
             `Can't upgrade from version ${storedVersion} to ${this.currentVersion}. Resetting database at ${this.dataDirectory}`,

yarn-project/validator-ha-signer/src/db/lmdb.test.ts

@@ -1,15 +1,16 @@
 import { BlockNumber, CheckpointNumber, IndexWithinCheckpoint, SlotNumber } from '@aztec/foundation/branded-types';
 import { EthAddress } from '@aztec/foundation/eth-address';
 import { TestDateProvider } from '@aztec/foundation/timer';
-import { type AztecLMDBStoreV2, openStoreAt, openTmpStore } from '@aztec/kv-store/lmdb-v2';
+import { type AztecLMDBStoreV2, createStore, openStoreAt, openTmpStore } from '@aztec/kv-store/lmdb-v2';
 
 import { afterEach, beforeEach, describe, expect, it } from '@jest/globals';
 import { mkdir, mkdtemp, rm } from 'fs/promises';
 import { tmpdir } from 'os';
 import { join } from 'path';
 
+import { createLocalSignerWithProtection } from '../factory.js';
 import { LmdbSlashingProtectionDatabase } from './lmdb.js';
-import { DutyStatus, DutyType } from './types.js';
+import { DutyStatus, DutyType, type StoredDutyRecord } from './types.js';
 
 describe('LmdbSlashingProtectionDatabase', () => {
   let store: AztecLMDBStoreV2;
@@ -418,3 +419,104 @@ describe('LmdbSlashingProtectionDatabase - persistence across restarts', () => {
     }
   });
 });
+
+describe('LmdbSlashingProtectionDatabase - schema migration', () => {
+  const ROLLUP_ADDRESS = EthAddress.random();
+  const VALIDATOR_ADDRESS = EthAddress.random();
+  const SLOT = SlotNumber(100);
+  const BLOCK_NUMBER = BlockNumber(50);
+  const BLOCK_INDEX = IndexWithinCheckpoint(0);
+  const DUTY_TYPE = DutyType.BLOCK_PROPOSAL;
+  const MESSAGE_HASH = '0xdeadbeef';
+  const NODE_ID = 'local';
+  const SIGNATURE = '0xsignature';
+
+  type StoredDutyRecordV1 = Omit<StoredDutyRecord, 'checkpointNumber'>;
+
+  let dataDir: string;
+  let dateProvider: TestDateProvider;
+
+  const defaultParams = () => ({
+    rollupAddress: ROLLUP_ADDRESS,
+    validatorAddress: VALIDATOR_ADDRESS,
+    slot: SLOT,
+    blockNumber: BLOCK_NUMBER,
+    checkpointNumber: CheckpointNumber(1),
+    blockIndexWithinCheckpoint: BLOCK_INDEX,
+    dutyType: DUTY_TYPE,
+    messageHash: MESSAGE_HASH,
+    nodeId: NODE_ID,
+  });
+
+  const createConfig = () => ({
+    l1Contracts: { rollupAddress: ROLLUP_ADDRESS },
+    nodeId: NODE_ID,
+    pollingIntervalMs: 100,
+    signingTimeoutMs: 3_000,
+    dataDirectory: dataDir,
+    dataStoreMapSizeKb: 1024 * 1024,
+  });
+
+  const seedSchemaVersion1Duty = async (record: StoredDutyRecordV1) => {
+    const store = await createStore('signing-protection', 1, createConfig());
+    const duties = store.openMap<string, StoredDutyRecordV1>('signing-protection-duties');
+    await duties.set(
+      `${record.rollupAddress}:${record.validatorAddress}:${record.slot}:${record.dutyType}:${record.blockIndexWithinCheckpoint}`,
+      record,
+    );
+    await store.close();
+  };
+
+  beforeEach(async () => {
+    dataDir = await mkdtemp(join(tmpdir(), 'lmdb-slashing-migration-'));
+    await mkdir(dataDir, { recursive: true });
+    dateProvider = new TestDateProvider();
+  });
+
+  afterEach(async () => {
+    await rm(dataDir, { recursive: true, force: true, maxRetries: 3, retryDelay: 100 });
+  });
+
+  it('migrates schema 1 duty records without allowing duplicate signing', async () => {
+    await seedSchemaVersion1Duty({
+      rollupAddress: ROLLUP_ADDRESS.toString(),
+      validatorAddress: VALIDATOR_ADDRESS.toString(),
+      slot: SLOT.toString(),
+      blockNumber: BLOCK_NUMBER.toString(),
+      blockIndexWithinCheckpoint: BLOCK_INDEX,
+      dutyType: DUTY_TYPE,
+      status: DutyStatus.SIGNED,
+      messageHash: MESSAGE_HASH,
+      signature: SIGNATURE,
+      nodeId: NODE_ID,
+      lockToken: 'legacy-lock-token',
+      startedAtMs: dateProvider.now(),
+      completedAtMs: dateProvider.now(),
+    });
+
+    const { db } = await createLocalSignerWithProtection(createConfig(), { dateProvider });
+    try {
+      const result = await db.tryInsertOrGetExisting(defaultParams());
+
+      expect(result.isNew).toBe(false);
+      expect(result.record.status).toBe(DutyStatus.SIGNED);
+      expect(result.record.signature).toBe(SIGNATURE);
+      expect(result.record.checkpointNumber).toBe(CheckpointNumber(0));
+    } finally {
+      await db.close();
+    }
+  });
+
+  it('fails closed instead of resetting when the stored schema is newer', async () => {
+    const store = await createStore(
+      'signing-protection',
+      LmdbSlashingProtectionDatabase.SCHEMA_VERSION + 1,
+      createConfig(),
+    );
+    await store.close();
+
+    await expect(createLocalSignerWithProtection(createConfig(), { dateProvider })).rejects.toThrow(
+      `stored schema version ${LmdbSlashingProtectionDatabase.SCHEMA_VERSION + 1} is incompatible with expected schema version ${LmdbSlashingProtectionDatabase.SCHEMA_VERSION}`,
+    );
+  });
+});

yarn-project/validator-ha-signer/src/db/lmdb.ts

@@ -13,6 +13,7 @@ import { EthAddress } from '@aztec/foundation/eth-address';
 import { type Logger, createLogger } from '@aztec/foundation/log';
 import type { DateProvider } from '@aztec/foundation/timer';
 import type { AztecAsyncKVStore, AztecAsyncMap } from '@aztec/kv-store';
+import { openStoreAt } from '@aztec/kv-store/lmdb-v2';
 
 import type { SlashingProtectionDatabase, TryInsertOrGetResult } from '../types.js';
 import {
@@ -24,6 +25,48 @@ import {
   recordFromFields,
 } from './types.js';
 
+const DUTIES_MAP_NAME = 'signing-protection-duties';
+const LEGACY_CHECKPOINT_NUMBER = '0';
+
+type StoredDutyRecordV1 = Omit<StoredDutyRecord, 'checkpointNumber'> & { checkpointNumber?: undefined };
+type MigratableStoredDutyRecord = StoredDutyRecord | StoredDutyRecordV1;
+
+function needsCheckpointNumberMigration(record: MigratableStoredDutyRecord): record is StoredDutyRecordV1 {
+  return record.checkpointNumber === undefined;
+}
+
+/**
+ * Migrates local slashing-protection duties from schema 1 to schema 2.
+ */
+export async function migrateLmdbSlashingProtectionDatabase(
+  dataDirectory: string,
+  currentVersion: number,
+  latestVersion: number,
+  dbMapSizeKb?: number,
+): Promise<void> {
+  if (currentVersion !== 1 || latestVersion !== LmdbSlashingProtectionDatabase.SCHEMA_VERSION) {
+    throw new Error(`Unsupported LMDB slashing-protection migration ${currentVersion} -> ${latestVersion}`);
+  }
+
+  const store = await openStoreAt(dataDirectory, dbMapSizeKb);
+  try {
+    const duties = store.openMap<string, MigratableStoredDutyRecord>(DUTIES_MAP_NAME);
+    const migratedRecords: { key: string; value: StoredDutyRecord }[] = [];
+
+    for await (const [key, record] of duties.entriesAsync()) {
+      if (needsCheckpointNumberMigration(record)) {
+        migratedRecords.push({ key, value: { ...record, checkpointNumber: LEGACY_CHECKPOINT_NUMBER } });
+      }
+    }
+
+    if (migratedRecords.length > 0) {
+      await duties.setMany(migratedRecords);
+    }
+  } finally {
+    await store.close();
+  }
+}
+
 function dutyKey(
   rollupAddress: string,
   validatorAddress: string,
@@ -51,7 +94,7 @@ export class LmdbSlashingProtectionDatabase implements SlashingProtectionDatabas
     private readonly dateProvider: DateProvider,
   ) {
     this.log = createLogger('slashing-protection:lmdb');
-    this.duties = store.openMap<string, StoredDutyRecord>('signing-protection-duties');
+    this.duties = store.openMap<string, StoredDutyRecord>(DUTIES_MAP_NAME);
   }
 
   /**

yarn-project/validator-ha-signer/src/factory.ts

@@ -8,7 +8,7 @@ import { getTelemetryClient } from '@aztec/telemetry-client';
 
 import { Pool } from 'pg';
 
-import { LmdbSlashingProtectionDatabase } from './db/lmdb.js';
+import { LmdbSlashingProtectionDatabase, migrateLmdbSlashingProtectionDatabase } from './db/lmdb.js';
 import { PostgresSlashingProtectionDatabase } from './db/postgres.js';
 import { HASignerMetrics } from './metrics.js';
 import type { CreateHASignerDeps, CreateLocalSignerWithProtectionDeps, SlashingProtectionDatabase } from './types.js';
@@ -119,11 +119,26 @@ export async function createLocalSignerWithProtection(
   const telemetryClient = deps?.telemetryClient ?? getTelemetryClient();
   const dateProvider = deps?.dateProvider ?? new DateProvider();
 
-  const kvStore = await createStore('signing-protection', LmdbSlashingProtectionDatabase.SCHEMA_VERSION, {
-    dataDirectory: config.dataDirectory,
-    dataStoreMapSizeKb: config.signingProtectionMapSizeKb ?? config.dataStoreMapSizeKb,
-    l1Contracts: config.l1Contracts,
-  });
+  const kvStore = await createStore(
+    'signing-protection',
+    LmdbSlashingProtectionDatabase.SCHEMA_VERSION,
+    {
+      dataDirectory: config.dataDirectory,
+      dataStoreMapSizeKb: config.signingProtectionMapSizeKb ?? config.dataStoreMapSizeKb,
+      l1Contracts: config.l1Contracts,
+    },
+    undefined,
+    {
+      onUpgrade: (dataDirectory, currentVersion, latestVersion) =>
+        migrateLmdbSlashingProtectionDatabase(
+          dataDirectory,
+          currentVersion,
+          latestVersion,
+          config.signingProtectionMapSizeKb ?? config.dataStoreMapSizeKb,
+        ),
+      schemaVersionMismatchPolicy: 'throw',
+    },
+  );
 
   const db = new LmdbSlashingProtectionDatabase(kvStore, dateProvider);