diff --git a/DESIGN.md b/DESIGN.md index 0df77c61b3..16913fe455 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -45,7 +45,7 @@ Index of the design notes for the harper core: one line per note, grouped by the - [Subscription version selection (`Table.ts`)](resources/DESIGN.md#subscription-version-selection-tablets) — Replay and live delivery share primary-version filtering, reading the current entry once per audit record rather than per subscriber; messages remain independent and raw events preserve history by default. - [Audit-store `'committed'` notification batching (`transactionBroadcast.ts`)](resources/DESIGN.md#audit-store-committed-notification-batching-transactionbroadcastts) — `'committed'` notifications are deferred with `setImmediate`, collapsed per turn and batched with yields; the ancestor key walk must strictly shrink; a subscription ended mid-delivery leaves its key's array to the loop walking it, and every delivery loop is bounded by what it started with. - [`createBlob(readable)` and `table.put()` don't synchronously drain the source](resources/DESIGN.md#createblobreadable-and-tableput-dont-synchronously-drain-the-source) — `table.put()` returns before a `Readable`-backed blob has drained; finalize a hash on the stream's end or await `storageInfo.saving`. -- [Table drops, the `dropping` tombstone, and ghost tables](resources/DESIGN.md#table-drops-the-dropping-tombstone-and-ghost-tables) — `dropTable()` persists a `dropping` tombstone before destructive work; boot and a same-name create complete an interrupted drop. +- [Table drops, the `dropping` tombstone, and ghost tables](resources/DESIGN.md#table-drops-the-dropping-tombstone-and-ghost-tables) — `dropTable()` persists a `dropping` tombstone before destructive work; boot and a same-name create complete an interrupted drop. First-time creates keep legacy names for rollback; recovery preserves names owned by a live table (harper#3102). - [The exclusive `update-attributes` lock is a bounded synchronous wait, and drop-then-recreate needs the column-family eviction fix (`Table.ts`)](resources/DESIGN.md#the-exclusive-update-attributes-lock-is-a-bounded-synchronous-wait-and-drop-then-recreate-needs-the-column-family-eviction-fix-tablets) — The lock is a bounded `Atomics.wait` with structural release, acquired before any live-`Table` mutation; drop-then-recreate needs the `@harperfast/rocksdb-js` column-family eviction fix. - [RocksDB transaction log purges are database-wide only (`ResourceBridge.deleteTransactionLogsBefore`)](resources/DESIGN.md#rocksdb-transaction-log-purges-are-database-wide-only-resourcebridgedeletetransactionlogsbefore) — RocksDB transaction logs are per database, so a table-scoped `delete_transaction_logs_before` is rejected with 400. - [Boot replay reads bypass the primary-record cache](resources/DESIGN.md#boot-replay-reads-bypass-the-primary-record-cache) — Snapshot reads avoid retaining the durable backlog through WeakRefs during synchronous boot replay. diff --git a/integrationTests/upgrade/first-create-downgrade.test.ts b/integrationTests/upgrade/first-create-downgrade.test.ts new file mode 100644 index 0000000000..6b97f18d27 --- /dev/null +++ b/integrationTests/upgrade/first-create-downgrade.test.ts @@ -0,0 +1,136 @@ +/** + * Tables first created after a minor upgrade remain readable and writable by 5.2, + * including secondary indexes and writes carried through a re-upgrade (#3102). + */ +import { suite, test, after } from 'node:test'; +import assert from 'node:assert'; +import { delimiter, join } from 'node:path'; +import { readFileSync } from 'node:fs'; +import { + startHarper, + killHarper, + teardownHarper, + sendOperation, + type ContextWithHarper, +} from '@harperfast/integration-testing'; + +const previousPath = process.env.HARPER_PREVIOUS_MINOR_PATH?.split(delimiter).find((path) => { + const { version } = JSON.parse(readFileSync(join(path, 'package.json'), 'utf8')); + return version.startsWith('5.2.'); +}); + +suite( + 'first-time table create survives 5.2 rollback', + { + skip: + !previousPath || + process.env.HARPER_RUNTIME === 'bun' || + process.env.HARPER_STORAGE_ENGINE === 'lmdb' || + process.platform === 'win32', + timeout: 300_000, + }, + (ctx: ContextWithHarper) => { + after(async () => teardownHarper(ctx)); + + test('preserves primary rows, secondary indexes and rollback writes', async () => { + const previousBin = join(previousPath!, 'dist', 'bin', 'harper.js'); + await startHarper(ctx, { config: {}, env: { TC_AGREEMENT: 'yes' }, harperBinPath: previousBin }); + await killHarper(ctx); + await startHarper(ctx, { config: {} }); + const schemas = [undefined, 'new_on_current']; + await sendOperation(ctx.harper, { operation: 'create_schema', schema: 'new_on_current' }); + const records = [ + { id: 'a', category: 'current' }, + { id: 'b', category: 'current' }, + ]; + for (const schema of schemas) { + await sendOperation(ctx.harper, { + operation: 'create_table', + schema, + table: 'born_on_current', + primary_key: 'id', + attributes: [ + { name: 'id', type: 'ID' }, + { name: 'category', type: 'String', indexed: true }, + ], + }); + await sendOperation(ctx.harper, { operation: 'upsert', schema, table: 'born_on_current', records }); + } + await sendOperation(ctx.harper, { operation: 'create_table', table: 'recreated', primary_key: 'id' }); + await sendOperation(ctx.harper, { operation: 'upsert', table: 'recreated', records: [{ id: 'retired' }] }); + await sendOperation(ctx.harper, { operation: 'drop_table', table: 'recreated' }); + await sendOperation(ctx.harper, { operation: 'create_table', table: 'recreated', primary_key: 'id' }); + await sendOperation(ctx.harper, { operation: 'upsert', table: 'recreated', records: [{ id: 'stamped' }] }); + await sendOperation(ctx.harper, { operation: 'create_table', table: 'rollback_recreate', primary_key: 'id' }); + await sendOperation(ctx.harper, { + operation: 'upsert', + table: 'rollback_recreate', + records: [{ id: 'retired' }], + }); + await sendOperation(ctx.harper, { operation: 'drop_table', table: 'rollback_recreate' }); + await killHarper(ctx); + await startHarper(ctx, { config: {}, env: { CONFIRM_DOWNGRADE: 'yes' }, harperBinPath: previousBin }); + await sendOperation(ctx.harper, { operation: 'create_table', table: 'rollback_recreate', primary_key: 'id' }); + await sendOperation(ctx.harper, { + operation: 'upsert', + table: 'rollback_recreate', + records: [{ id: 'legacy-recreated' }], + }); + + const read = (schema: string | undefined, attribute = 'id', value = '*') => + sendOperation(ctx.harper, { + operation: 'search_by_value', + schema, + table: 'born_on_current', + search_attribute: attribute, + search_value: value, + get_attributes: ['id', 'category'], + }); + assert.deepStrictEqual( + await sendOperation(ctx.harper, { + operation: 'search_by_value', + table: 'recreated', + search_attribute: 'id', + search_value: '*', + get_attributes: ['*'], + }), + [], + '5.2 does not support generation-stamped recreates' + ); + const rollbackRecord = { id: 'c', category: 'rollback' }; + for (const schema of schemas) { + assert.deepStrictEqual(await read(schema), records); + assert.deepStrictEqual(await read(schema, 'category', 'current'), records); + await sendOperation(ctx.harper, { + operation: 'upsert', + schema, + table: 'born_on_current', + records: [rollbackRecord], + }); + } + await killHarper(ctx); + await startHarper(ctx, { config: {} }); + const readRecreated = () => + sendOperation(ctx.harper, { + operation: 'search_by_value', + table: 'rollback_recreate', + search_attribute: 'id', + search_value: '*', + get_attributes: ['id'], + }); + assert.deepStrictEqual(await readRecreated(), [{ id: 'legacy-recreated' }]); + await killHarper(ctx); + await startHarper(ctx, { config: {} }); + assert.deepStrictEqual( + await readRecreated(), + [{ id: 'legacy-recreated' }], + 'a retired journal must not reclaim the bare table recreated on 5.2' + ); + for (const schema of schemas) { + assert.deepStrictEqual(await read(schema), [...records, rollbackRecord]); + assert.deepStrictEqual(await read(schema, 'category', 'current'), records); + assert.deepStrictEqual(await read(schema, 'category', 'rollback'), [rollbackRecord]); + } + }); + } +); diff --git a/resources/DESIGN.md b/resources/DESIGN.md index 4ead587170..6ec3b1c895 100644 --- a/resources/DESIGN.md +++ b/resources/DESIGN.md @@ -617,6 +617,8 @@ interrupted drop instead of resurrecting the table. Without this, surviving cata silently re-opened with create-if-missing on the next start, which resurrects "deleted" tables (with their data, if the column families were never actually removed). +**First-time creates and 5.2 rollback (harper#3102).** First-time RocksDB creates use bare primary/index names only when this node has no `/dropped/` row, no journal row for the table, and no `T/` column family. Every RocksDB drop completion (`retireRocksStores` in `Table.ts`, `completeInterruptedDrop`) writes that untimed name-history row (`recordTableNameHistory`) before removing the tombstone, so any name history requires a generation stamp, including after reclamation. The load parser skips `/dropped/` rows; 5.4 reads the same row as its untimed drop marker (harper#2962), which a timed drop overwrites. Incoming primary-attribute generations are replaced by this node's physical naming choice. Create journals have their own identity; when a creating and retired journal name the same primary, the creating row leaves that primary to the retired row's blob sweep. Recovery reads their current phases under the catalog lock. Their `creatingStores` field must stay separate from `stores`, which shipped 5.3 readers destructively reclaim without recognizing a published bare primary. Reclamation always preserves stores owned by the live, non-dropping catalog and never sweeps that primary's blobs: a 5.2 writer can reuse bare names while ignoring the retired journal. Previously stamped tables retain their names and remain outside 5.2 rollback support, as do stamped recreates. During interrupted-drop recovery, a bare predecessor can remain until its asynchronous reclamation finishes; 5.2 ignores that recovery state and can expose predecessor rows through a recreate. `unitTests/resources/dropTableGeneration.test.js` enforces naming, crash recovery, and live-store ownership; `integrationTests/upgrade/first-create-downgrade.test.ts` exercises the real 5.2 round trip. + ## The exclusive `update-attributes` lock is a bounded synchronous wait, and drop-then-recreate needs the column-family eviction fix (`Table.ts`) The create/schema-update path's exclusive `update-attributes` lock is a synchronous bounded wait diff --git a/resources/Table.ts b/resources/Table.ts index ca3f108932..d1b4d9792d 100644 --- a/resources/Table.ts +++ b/resources/Table.ts @@ -88,6 +88,8 @@ import { sweepDroppedTableBlobs, storeNameFor, storeNamesFor, + liveStoreNamesFor, + recordTableNameHistory, isReadOnlyMode, } from './databases.ts'; import { notifyReplicatedApplyFailure } from './replicatedApplyFailure.ts'; @@ -3131,6 +3133,13 @@ export function makeTable(options): TableResourceClass { // has stopped and the wrapper has retired their storage. if (!(await retireFullTextStorage())) return false; const removed = withUpdateAttributesLock(rootStore, `table '${databaseName}.${tableName}'`, () => { + const currentPrimary = (dbisDb as any).getSync(primaryCatalogKey); + if ( + !currentPrimary?.dropping || + (currentPrimary.tableId != null && currentPrimary.tableId !== tableId) || + currentPrimary.dropGeneration !== dropGeneration + ) + return false; const stores = storeNamesFor(dbisDb, tableName, generation); const retiredStores = recordRetiredGeneration( dbisDb, @@ -3164,25 +3173,35 @@ export function makeTable(options): TableResourceClass { if (!droppedStores.has(columnName) && (rootStore as any).columns.includes(columnName)) dropColumnFamily(rootStore, columnName); } - const currentPrimary = (dbisDb as any).getSync(primaryCatalogKey); + const remainingPrimary = (dbisDb as any).getSync(primaryCatalogKey); if ( - !currentPrimary?.dropping || - (currentPrimary.tableId != null && currentPrimary.tableId !== tableId) || - currentPrimary.dropGeneration !== dropGeneration + !remainingPrimary?.dropping || + (remainingPrimary.tableId != null && remainingPrimary.tableId !== tableId) || + remainingPrimary.dropGeneration !== dropGeneration ) return false; for (const key of dbisDb.getKeys({ start: tableName + '/', end: tableName + '0' })) { if (key !== primaryCatalogKey) dbisDb.remove(key); } + recordTableNameHistory(dbisDb, tableName, currentPrimary.tableId); dbisDb.remove(primaryCatalogKey); return true; }); - if (removed) await dbisDb.committed; + if (!removed) { + withUpdateAttributesLock(rootStore, `retire stale stores of '${databaseName}.${tableName}'`, () => { + const owned = liveStoreNamesFor(dbisDb, tableName); + const columns = new Set((rootStore as any).columns); + for (const store of [primaryStore, ...Object.values(indices)]) { + if (store.name && columns.has(store.name) && !owned.has(store.name)) + dropColumnFamily(rootStore, store.name); + } + }); + } else await dbisDb.committed; const label = `${databaseName}.${tableName}`; const settled = await settlePhysicalDrops(rootStore, label); await sweepDroppedTableBlobs(primaryStore, label); if (!settled) finishDroppedTableBlobSweep(rootStore, primaryStore, label); - return true; + return removed; } finally { releaseDropMark(); } diff --git a/resources/databases.ts b/resources/databases.ts index 0ad7821877..9806c7c40f 100644 --- a/resources/databases.ts +++ b/resources/databases.ts @@ -595,6 +595,7 @@ type GenerationRow = { generation: string; phase: 'creating' | 'retired'; stores?: string[]; + creatingStores?: string[]; primaryStore?: string; blobSweepFailures?: number; }; @@ -649,6 +650,20 @@ export function recordRetiredGeneration( }); return merged; } +const DROPPED_ROW_PREFIX = '/dropped/'; +function droppedRowKey(tableName: string): string { + return DROPPED_ROW_PREFIX + tableName; +} +/** + * Once a name has been dropped, later creates of it take a generation even after the retirement journal is + * reclaimed. Keep the row untimed and shaped `{ table, tableId }`: later versions read it as an untimed drop + * marker that a timed drop overwrites. RocksDB callers hold the catalog lock and write it before removing + * the tombstone. + */ +export function recordTableNameHistory(attributesDbi, tableName: string, tableId: number | undefined): void { + const key = droppedRowKey(tableName); + if (!attributesDbi.getSync(key)) attributesDbi.putSync(key, { table: tableName, tableId }); +} export function storeNamesFor(attributesDbi, tableName: string, generation: string | undefined): string[] { const names: string[] = []; for (const key of attributesDbi.getKeys({ start: tableName + '/', end: tableName + '0' })) { @@ -1292,7 +1307,8 @@ function initStores( for (const result of attributesDbi.getRange({ start: false })) { const { key, value } = result as { key: string; value: any }; if (value == null) continue; - if (typeof key === 'string' && key.startsWith(GENERATION_ROW_PREFIX)) continue; + if (typeof key === 'string' && (key.startsWith(GENERATION_ROW_PREFIX) || key.startsWith(DROPPED_ROW_PREFIX))) + continue; let [tableName, attribute_name] = key.toString().split('/'); if (attribute_name === '') { // primary key @@ -3635,6 +3651,7 @@ function declareTable(target: TableTarget, tableDefinition: T let refreshedLiveAttributes = false; let deferredPrimaryRow: any; let generation: string | undefined; + let createJournal: GenerationRow | undefined; let unpublishedPrimaryStore: any; let published = false; let fullTextValuesForPersistence: unknown; @@ -4177,8 +4194,33 @@ function declareTable(target: TableTarget, tableDefinition: T clearInterruptedDropEntries(rootStore.path, tableName); } if (rootStore instanceof RocksDatabase) { - generation = randomUUID(); - attributesDbi.putSync(generationRowKey(generation), { table: tableName, generation, phase: 'creating' }); + const journalId = randomUUID(); + let hasPriorStores = Boolean( + attributesDbi.getSync(droppedRowKey(tableName)) || rootStore.columns.some((name) => name.startsWith(dbiName)) + ); + if (!hasPriorStores) { + for (const { value } of attributesDbi.getRange({ start: GENERATION_ROW_PREFIX, end: GENERATION_ROW_END })) { + if (value?.table === tableName) { + hasPriorStores = true; + break; + } + } + } + if (hasPriorStores) generation = journalId; + createJournal = { + table: tableName, + generation: journalId, + phase: 'creating', + primaryStore: storeNameFor(dbiName, generation), + // Shipped 5.3 readers reclaim `stores` even when a bare primary row was published. + creatingStores: [ + storeNameFor(dbiName, generation), + ...attributes + .filter((attribute) => attribute.indexed && !attribute.isPrimaryKey && !attribute.relationship) + .map((attribute) => storeNameFor(`${dbiName}${attribute.name}`, generation)), + ], + }; + attributesDbi.putSync(generationRowKey(journalId), createJournal); primaryStore = openRocksDatabase(rootStore.path, { ...dbiInit, name: storeNameFor(dbiName, generation), @@ -4200,6 +4242,7 @@ function declareTable(target: TableTarget, tableDefinition: T primaryKeyAttribute.tableId = primaryStore.tableId; if (generation) primaryKeyAttribute.generation = generation; + else delete primaryKeyAttribute.generation; Table = makeTable({ isBranch: Boolean(target.branch), primaryStore, @@ -4687,13 +4730,13 @@ function declareTable(target: TableTarget, tableDefinition: T // below is a no-op for a create — a table is never published with an incomplete relationship list. if (deferredPrimaryRow) { attributesDbi.put(tableName + '/', deferredPrimaryRow); - if (generation) attributesDbi.remove(generationRowKey(generation)); // That write, not the registration below, is the publish point: it is durable from here // (on LMDB releaseLock()'s finally commits this create's write transaction even while an // error unwinds), so any later throw must leave the catalog alone. Rolling back past it // would delete the attribute rows out from under a live primary row and leave every // thread loading the primary-only schema this change exists to prevent. published = true; + if (createJournal) attributesDbi.remove(generationRowKey(createJournal.generation)); setTable(tables, tableName, Table); } // a table with no declared primary key has no attribute row to carry relationships, and the @@ -4806,14 +4849,17 @@ function declareTable(target: TableTarget, tableDefinition: T } } function discardUnpublishedTable() { + let storeDropFailed = false; const discard = (description: string, action: () => unknown) => { try { action(); + return true; } catch (discardError) { logger.warn( `Error discarding ${description} of the failed create of ${databaseName}.${tableName}`, discardError ); + return false; } }; discard('catalog rows', () => { @@ -4825,13 +4871,15 @@ function declareTable(target: TableTarget, tableDefinition: T // an LMDB store is a per-environment handle slot shared with every thread and still inside this // create's write transaction; only RocksDB column-family handles hold native state to release if (rootStore instanceof RocksDatabase) { - if (generation) { - const suffix = '@' + generation; + if (createJournal) { + const suffix = '@' + createJournal.generation; + const stores = new Set(createJournal.creatingStores); for (const columnName of [...((rootStore as any).columns as string[])]) { - if (columnName.endsWith(suffix)) - discard(`store ${columnName}`, () => dropColumnFamily(rootStore, columnName)); + if (stores.has(columnName) || columnName.endsWith(suffix)) + if (!discard(`store ${columnName}`, () => dropColumnFamily(rootStore, columnName))) storeDropFailed = true; } - discard('generation journal row', () => attributesDbi.remove(generationRowKey(generation))); + if (!storeDropFailed) + discard('generation journal row', () => attributesDbi.remove(generationRowKey(createJournal.generation))); } for (const indexName in Table?.indices ?? {}) discard(`index ${indexName}`, () => Table.indices[indexName].close()); @@ -5398,6 +5446,20 @@ export async function sweepDroppedTableBlobs( return { failures, cancelled: Boolean(cancelled()), batches }; } +/** Older writers can reuse bare names without updating their retired journals. */ +export function liveStoreNamesFor(attributesDbi, tableName: string): Set { + let primary = attributesDbi.getSync(tableName + '/'); + if (!primary) { + for (const { value } of attributesDbi.getRange({ start: tableName + '/', end: tableName + '0' })) { + if (value?.isPrimaryKey || value?.is_hash_attribute) { + primary = value; + break; + } + } + } + return new Set(primary && !primary.dropping ? storeNamesFor(attributesDbi, tableName, primary.generation) : []); +} + function reclaimGenerations(rootStore: RocksDatabase, attributesDbi, databaseName: string) { const rows = Array.from(attributesDbi.getRange({ start: GENERATION_ROW_PREFIX, end: GENERATION_ROW_END })); const state = scheduledGenerationReclaims.get(rootStore); @@ -5415,7 +5477,18 @@ function reclaimGenerations(rootStore: RocksDatabase, attributesDbi, databaseNam return; } try { - for (const { key, value } of rows as Array<{ key: string; value: GenerationRow }>) { + const currentRows = Array.from( + attributesDbi.getRange({ start: GENERATION_ROW_PREFIX, end: GENERATION_ROW_END }) + ) as Array<{ key: string; value: GenerationRow }>; + const retiredPrimaries = new Set( + currentRows + .filter( + ({ value }) => + value?.phase === 'retired' && value.generation && value.table && typeof value.primaryStore === 'string' + ) + .map(({ value }) => value.primaryStore) + ); + for (const { key, value } of currentRows) { try { if (!value?.generation || !value.table) { logger.warn(`Removing a malformed generation journal row ${String(key)} in ${databaseName}`); @@ -5424,21 +5497,36 @@ function reclaimGenerations(rootStore: RocksDatabase, attributesDbi, databaseNam continue; } const live = attributesDbi.getSync(value.table + '/'); - if (value.phase === 'creating' && live?.generation === value.generation && !live.dropping) { + const publishedCreate = + live && + (typeof value.primaryStore === 'string' + ? storeNameFor(value.table + '/', live.generation) === value.primaryStore + : live.generation === value.generation); + if (value.phase === 'creating' && publishedCreate) { attributesDbi.remove(key); resetGenerationReclaimDelay(rootStore); continue; } const suffix = '@' + value.generation; - const retired = new Set(value.stores ?? []); + const retired = new Set( + value.phase === 'creating' ? (value.creatingStores ?? value.stores ?? []) : (value.stores ?? []) + ); const columns = [...((rootStore as any).columns as string[])]; - if (value.phase === 'retired' && value.primaryStore && columns.includes(value.primaryStore)) { + const owned = liveStoreNamesFor(attributesDbi, value.table); + if ( + value.phase === 'retired' && + value.primaryStore && + columns.includes(value.primaryStore) && + !owned.has(value.primaryStore) + ) { if (manageThreads.ownsStoreMaintenance(rootStore.path)) scheduleGenerationBlobSweep(rootStore, attributesDbi, databaseName, key, value); continue; } for (const columnName of columns) { - if (retired.has(columnName) || columnName.endsWith(suffix)) dropColumnFamily(rootStore, columnName); + if (value.phase === 'creating' && retiredPrimaries.has(columnName)) continue; + if (!owned.has(columnName) && (retired.has(columnName) || columnName.endsWith(suffix))) + dropColumnFamily(rootStore, columnName); } if ((rootStore.getStats?.()?.['columnFamily.pendingReclaims'] ?? 0) > 0) { scheduleGenerationReclaim(rootStore, attributesDbi, databaseName); @@ -5493,8 +5581,10 @@ async function finishGenerationBlobSweep( } const suffix = '@' + latest.generation; const retired = new Set(latest.stores ?? []); + const owned = liveStoreNamesFor(attributesDbi, latest.table); for (const columnName of [...((rootStore as any).columns as string[])]) { - if (retired.has(columnName) || columnName.endsWith(suffix)) dropColumnFamily(rootStore, columnName); + if (!owned.has(columnName) && (retired.has(columnName) || columnName.endsWith(suffix))) + dropColumnFamily(rootStore, columnName); } if ((rootStore.getStats?.()?.['columnFamily.pendingReclaims'] ?? 0) === 0) { attributesDbi.remove(key); @@ -5523,7 +5613,8 @@ function scheduleGenerationBlobSweep( current?.phase !== 'retired' || current.generation !== row.generation || current.primaryStore !== row.primaryStore || - !((rootStore as any).columns as string[]).includes(row.primaryStore) + !((rootStore as any).columns as string[]).includes(row.primaryStore) || + liveStoreNamesFor(attributesDbi, row.table).has(row.primaryStore) ) return; let result: Awaited>; @@ -5730,7 +5821,11 @@ function completeInterruptedDrop( // catalog-removal failure would bypass the retry accounting entirely. (attributesDbi as any).removeSync(key); } - if (tombstoneEntry) (attributesDbi as any).removeSync(tombstoneEntry.key); + if (tombstoneEntry) { + if (rootStore instanceof RocksDatabase) + recordTableNameHistory(attributesDbi, tableName, tombstoneEntry.value?.tableId); + (attributesDbi as any).removeSync(tombstoneEntry.key); + } return true; } diff --git a/unitTests/resources/dropTableGeneration.test.js b/unitTests/resources/dropTableGeneration.test.js index 195b88f693..6385082b5b 100644 --- a/unitTests/resources/dropTableGeneration.test.js +++ b/unitTests/resources/dropTableGeneration.test.js @@ -3,6 +3,7 @@ require('../testUtils'); const assert = require('node:assert'); const fs = require('node:fs'); +const { randomUUID } = require('node:crypto'); const { setupTestDBPath } = require('../testUtils'); const { waitFor } = require('../waitFor'); const { @@ -12,9 +13,11 @@ const { resetDatabases, openRocksDatabase, setDroppedBlobSweepBatchMsForTesting, + markDropInProgress, } = require('#src/resources/databases'); const { createBlob, getFilePathForBlob } = require('#src/resources/blob'); const { logger } = require('#src/utility/logging/logger'); +const { forComponent } = require('#src/utility/logging/harper_logger'); const { getPlaneBinding } = require('#src/resources/indexes/hnswPlaneBinding'); const { derivedIndexReadiness } = require('#src/resources/indexes/hnswDerivedIndex'); const { setMainIsWorker } = require('#js/server/threads/manageThreads'); @@ -89,8 +92,15 @@ describe('dropTable generation-distinct stores', function () { setMainIsWorker(true); }); - it('stamps every physical store of a RocksDB table with the generation on its catalog row', async function () { + it('keeps first-time creates readable by legacy readers and stamps every store of a recreate', async function () { if (IS_LMDB) return this.skip(); + const First = defineTable('GenStamped'); + assert.equal(dbisDb().getSync('GenStamped/').generation, undefined); + assert.equal(First.primaryStore.name, 'GenStamped/'); + assert.equal(First.indices.str.name, 'GenStamped/str'); + await First.dropTable(); + resetDatabases(); + assert.deepStrictEqual(generationRows(), [], 'the recreate is stamped even after retirement finishes'); const Stamped = defineTable('GenStamped'); const generation = dbisDb().getSync('GenStamped/').generation; assert.match(generation, /^[0-9a-f-]{36}$/, 'the primary row carries a create-time generation'); @@ -110,6 +120,69 @@ describe('dropTable generation-distinct stores', function () { await Plain.dropTable(); }); + it('keeps a /dropped/ name-history row after reclamation', async function () { + if (IS_LMDB) return this.skip(); + const First = defineTable('GenLocalDrop'); + assert.equal(First.storageGeneration, undefined); + await First.dropTable(); + resetDatabases(); + assert.ok(dbisDb().getSync('/dropped/GenLocalDrop')); + assert.deepStrictEqual(generationRows(), []); + const Fresh = defineTable('GenLocalDrop'); + assert.match(Fresh.storageGeneration, /^[0-9a-f-]{36}$/); + assert.notEqual(Fresh.primaryStore.name, First.primaryStore.name); + await Fresh.dropTable(); + }); + + it('persists the local store name when a peer primary attribute carries a generation', async function () { + const PeerDefined = table({ + database: TEST_DB, + table: 'GenPeerStamp', + origin: 'cluster', + attributes: [{ name: 'id', isPrimaryKey: true, generation: randomUUID() }], + }); + assert.equal(dbisDb().getSync('GenPeerStamp/').generation, undefined); + await PeerDefined.put({ id: 1 }); + resetDatabases(); + const Reloaded = databases[TEST_DB].GenPeerStamp; + assert.equal((await Reloaded.get(1)).id, 1); + await Reloaded.dropTable(); + }); + + it('does not retire catalog stores after a schema listener completes the drop and recreates the table', async function () { + if (IS_LMDB) return this.skip(); + const First = defineTable('GenListenerRecreate', [{ name: 'blob', type: 'Blob' }]); + const blob = await createBlob(Buffer.alloc(50_000, 4)); + await First.put({ id: 1, str: 'retired', blob }); + const blobPath = getFilePathForBlob((await First.get(1)).blob); + let Fresh; + const removeListener = schemaHandler.addListener((message) => { + if (message.operation === 'drop_table' && message.table === First.tableName && !Fresh) + Fresh = defineTable(First.tableName, [{ name: 'replacementOnly', type: 'String', indexed: true }]); + }); + try { + await First.dropTable(); + } finally { + removeListener(); + } + assert.ok(Fresh, 'the recreate runs before the old drop resumes from its broadcast'); + assert.ok(!rootStore().columns.includes(First.primaryStore.name), 'the stale drop retires its bare primary'); + assert.ok(!rootStore().columns.includes(First.indices.str.name), 'the stale drop retires its bare index'); + await waitFor(() => !fs.existsSync(blobPath), { timeout: 15_000 }); + await Fresh.put({ id: 1, str: 'replacement' }); + assert.equal((await Fresh.get(1)).str, 'replacement'); + assert.ok(!generationRows().some(({ value }) => value.stores?.includes(Fresh.primaryStore.name))); + assert.ok( + !generationRows().some( + ({ value }) => + value.table === Fresh.tableName && + value.stores?.some((name) => name.startsWith(`${Fresh.tableName}/replacementOnly`)) + ), + 'the resumed drop must not journal the replacement catalog' + ); + await Fresh.dropTable(); + }); + it('serves nothing from a dropped generation through a same-name recreate, primary or index', async function () { const First = defineTable('GenRecreate'); await First.put({ id: 1, str: 'original' }); @@ -150,15 +223,47 @@ describe('dropTable generation-distinct stores', function () { ); }); + it('preserves bare stores and blobs recreated by a legacy writer despite a retired journal', async function () { + if (IS_LMDB) return this.skip(); + const First = defineTable('GenLegacyRecreate', [{ name: 'oldOnly', type: 'String', indexed: true }]); + await First.put({ id: 1, str: 'retired' }); + await First.dropTable(); + const retired = generationRows().find(({ value }) => value.table === First.tableName); + assert.ok(retired); + // 5.2 ignores the drop marker and journal when it chooses physical names. + dbisDb().removeSync(retired.key); + dbisDb().removeSync('/dropped/' + First.tableName); + const Fresh = defineTable(First.tableName, [{ name: 'blob', type: 'Blob' }]); + assert.equal(Fresh.primaryStore.name, First.tableName + '/'); + const blob = await createBlob(Buffer.alloc(50_000, 3)); + await Fresh.put({ id: 2, str: 'rollback', blob }); + const blobPath = getFilePathForBlob((await Fresh.get(2)).blob); + openRocksDatabase(rootStore().path, { name: First.tableName + '/oldOnly' }).close(); + dbisDb().putSync(retired.key, retired.value); + resetDatabases(); + await waitFor(() => !dbisDb().getSync(retired.key), { timeout: 15_000 }); + const Reloaded = databases[TEST_DB][First.tableName]; + assert.equal((await Reloaded.get(2)).str, 'rollback'); + assert.deepStrictEqual( + (await fromAsync(Reloaded.search({ conditions: [{ attribute: 'str', value: 'rollback' }] }))).map(({ id }) => id), + [2] + ); + assert.ok(fs.existsSync(blobPath), 'reclaiming an old journal must not unlink a live blob'); + assert.ok( + !rootStore().columns.includes(First.tableName + '/oldOnly'), + 'unowned retired stores still get reclaimed' + ); + await Reloaded.dropTable(); + }); + it('keeps the retirement journal complete under a redundant concurrent drop', async function () { if (IS_LMDB) return this.skip(); const Twice = defineTable('GenDoubleDrop'); await Twice.put({ id: 1, str: 'x' }); - const { generation } = dbisDb().getSync('GenDoubleDrop/'); const family = Twice.primaryStore.name; await Promise.all([Twice.dropTable(), Twice.dropTable()]); assert.deepStrictEqual(catalogRows('GenDoubleDrop'), []); - const journal = dbisDb().getSync(`${GENERATION_ROW_PREFIX}${generation}`); + const journal = generationRows().find(({ value }) => value.table === 'GenDoubleDrop')?.value; assert.ok(journal, 'the retirement row stays until a load confirms the reclaim'); assert.ok(journal.stores.includes(family), 'the second drop must not narrow the store list'); resetDatabases(); @@ -180,7 +285,6 @@ describe('dropTable generation-distinct stores', function () { it('does not dispose a same-name replacement for a delayed old-generation drop event', async function () { const First = defineTable('GenDelayedDrop'); - const oldGeneration = First.storageGeneration; const oldTableId = First.tableId; let broadcast; const removeListener = schemaHandler.addListener((message) => { @@ -193,7 +297,7 @@ describe('dropTable generation-distinct stores', function () { removeListener(); } assert.equal(broadcast.dropTableId, oldTableId); - if (!IS_LMDB) assert.equal(broadcast.dropGeneration, oldGeneration); + if (!IS_LMDB) assert.match(broadcast.dropGeneration, /^[0-9a-f-]{36}$/); const Replacement = defineTable('GenDelayedDrop'); await Replacement.put({ id: 1, str: 'replacement' }); await schemaHandler({ @@ -213,7 +317,7 @@ describe('dropTable generation-distinct stores', function () { operation: 'drop_table', schema: TEST_DB, table: 'GenDelayedDrop', - dropGeneration: oldGeneration, + dropGeneration: broadcast.dropGeneration, dropTableId: oldTableId, }, }); @@ -292,7 +396,7 @@ describe('dropTable generation-distinct stores', function () { await Doomed.put({ id, str: 'old', blob }); blobPaths.push(getFilePathForBlob((await Doomed.get(id)).blob)); } - const { generation } = dbisDb().getSync('GenCrashRetired/'); + const generation = randomUUID(); const family = Doomed.primaryStore.name; // the drop died after writing its journal row and removing the catalog rows, before the // physical drop landed: the family is still on disk under its name @@ -300,6 +404,7 @@ describe('dropTable generation-distinct stores', function () { table: 'GenCrashRetired', generation, phase: 'retired', + stores: [family, Doomed.indices.str.name], primaryStore: family, }); for (const key of catalogRows('GenCrashRetired')) dbisDb().removeSync(key); @@ -349,12 +454,145 @@ describe('dropTable generation-distinct stores', function () { assert.deepStrictEqual(generationRows(), []); }); + it('reclaims primary and index stores of an unpublished legacy create', function () { + const generation = randomUUID(); + const stores = ['GenBareCrash/', 'GenBareCrash/str']; + dbisDb().putSync(`${GENERATION_ROW_PREFIX}${generation}`, { + table: 'GenBareCrash', + generation, + phase: 'creating', + primaryStore: stores[0], + creatingStores: stores, + }); + for (const name of stores) { + const orphan = openRocksDatabase(rootStore().path, { name }); + orphan.putSync(1, { id: 1 }); + orphan.close(); + } + resetDatabases(); + assert.ok(stores.every((name) => !rootStore().columns.includes(name))); + assert.deepStrictEqual(generationRows(), []); + }); + + it('preserves a published legacy create whose journal removal was interrupted', async function () { + const Published = defineTable('GenBarePublished'); + assert.equal(Published.primaryStore.name, 'GenBarePublished/'); + await Published.put({ id: 1, str: 'published' }); + const generation = randomUUID(); + const row = { + table: Published.tableName, + generation, + phase: 'creating', + primaryStore: Published.primaryStore.name, + creatingStores: [Published.primaryStore.name, Published.indices.str.name], + }; + dbisDb().putSync(`${GENERATION_ROW_PREFIX}${generation}`, row); + resetDatabases(); + const Reloaded = databases[TEST_DB].GenBarePublished; + assert.equal((await Reloaded.get(1)).str, 'published'); + assert.deepStrictEqual(generationRows(), []); + await Reloaded.dropTable(); + }); + + it('leaves a published legacy family to the dropper when its create journal survived publication', async function () { + const Dropping = defineTable('GenPublishedDropping'); + await Dropping.put({ id: 1, str: 'owned by the dropper' }); + const generation = randomUUID(); + dbisDb().putSync(`${GENERATION_ROW_PREFIX}${generation}`, { + table: Dropping.tableName, + generation, + phase: 'creating', + primaryStore: Dropping.primaryStore.name, + creatingStores: [Dropping.primaryStore.name, Dropping.indices.str.name], + }); + const primary = dbisDb().getSync(`${Dropping.tableName}/`); + primary.dropping = true; + primary.dropGeneration = randomUUID(); + dbisDb().putSync(`${Dropping.tableName}/`, primary); + const releaseDrop = markDropInProgress(primary.dropGeneration); + try { + resetDatabases(); + assert.ok( + rootStore().columns.includes(Dropping.primaryStore.name), + 'create recovery must not retire a live drop' + ); + assert.equal(Dropping.primaryStore.getSync(1).str, 'owned by the dropper'); + } finally { + releaseDrop(); + resetDatabases(); + } + }); + + for (const creatingFirst of [true, false]) { + it(`sweeps an interrupted drop before reclaiming its surviving create journal (${creatingFirst ? 'creating' : 'retired'} first)`, async function () { + const Doomed = defineTable(`GenOverlap${creatingFirst}`, [{ name: 'blob', type: 'Blob' }]); + assert.equal(Doomed.storageGeneration, undefined); + const blob = await createBlob(Buffer.alloc(50_000, creatingFirst ? 5 : 6)); + await Doomed.put({ id: 1, blob }); + const blobPath = getFilePathForBlob((await Doomed.get(1)).blob); + const ids = ['11111111-1111-4111-8111-111111111111', 'eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee']; + const [creatingId, retiredId] = creatingFirst ? ids : ids.toReversed(); + dbisDb().putSync(`${GENERATION_ROW_PREFIX}${creatingId}`, { + table: Doomed.tableName, + generation: creatingId, + phase: 'creating', + primaryStore: Doomed.primaryStore.name, + creatingStores: [Doomed.primaryStore.name, Doomed.indices.str.name], + }); + const primary = dbisDb().getSync(`${Doomed.tableName}/`); + primary.dropping = true; + primary.dropGeneration = retiredId; + dbisDb().putSync(`${Doomed.tableName}/`, primary); + resetDatabases(); + await waitFor( + () => + !dbisDb().getSync(`${GENERATION_ROW_PREFIX}${creatingId}`) && + !dbisDb().getSync(`${GENERATION_ROW_PREFIX}${retiredId}`), + { timeout: 15_000 } + ); + assert.ok(!fs.existsSync(blobPath), 'journal completion must follow blob unlinking'); + assert.ok(!rootStore().columns.includes(Doomed.primaryStore.name)); + }); + } + + it('keeps name history from a drop completed at load after its journal is reclaimed', async function () { + const Doomed = defineTable('GenLoadCompleted'); + assert.equal(Doomed.storageGeneration, undefined); + const primary = dbisDb().getSync('GenLoadCompleted/'); + primary.dropping = true; + primary.dropGeneration = randomUUID(); + dbisDb().putSync('GenLoadCompleted/', primary); + resetDatabases(); + await waitFor(() => !generationRows().some(({ value }) => value.table === 'GenLoadCompleted'), { + timeout: 15_000, + }); + assert.deepStrictEqual(catalogRows('GenLoadCompleted'), []); + assert.ok(dbisDb().getSync('/dropped/GenLoadCompleted')); + const storageLogger = forComponent('storage'); + const { warn, debug } = storageLogger; + const skipped = []; + // the first skip of a keyless catalog warns, a repeat logs at debug + storageLogger.warn = storageLogger.debug = (message) => { + if (String(message).startsWith('Skipping table')) skipped.push(message); + }; + try { + resetDatabases(); + } finally { + storageLogger.warn = warn; + storageLogger.debug = debug; + } + assert.deepStrictEqual(skipped, [], 'the load parser skips name-history rows'); + const Fresh = defineTable('GenLoadCompleted'); + assert.match(Fresh.storageGeneration, /^[0-9a-f-]{36}$/); + await Fresh.dropTable(); + }); + it('journals an interrupted legacy catalog whose primary row carries its attribute name', async function () { const Legacy = defineTable('GenLegacyPrimary'); const primary = dbisDb().getSync('GenLegacyPrimary/'); - const { generation } = primary; + const generation = randomUUID(); const primaryKey = 'GenLegacyPrimary/id'; - const family = `${primaryKey}@${generation}`; + const family = primaryKey; Legacy.primaryStore.dropSync(); const legacyStore = openRocksDatabase(rootStore().path, { name: family }); legacyStore.putSync(1, { id: 1, str: 'legacy' }); @@ -379,9 +617,9 @@ describe('dropTable generation-distinct stores', function () { it('keeps a migrated bare tombstone until after its named primary descriptor is removed', async function () { const Migrated = defineTable('GenMigratedPrimary'); const bare = dbisDb().getSync('GenMigratedPrimary/'); - const { generation } = bare; + const generation = randomUUID(); const primaryKey = 'GenMigratedPrimary/id'; - const family = `${primaryKey}@${generation}`; + const family = primaryKey; Migrated.primaryStore.dropSync(); const migratedStore = openRocksDatabase(rootStore().path, { name: family }); migratedStore.putSync(1, { id: 1, str: 'migrated' }); @@ -407,6 +645,7 @@ describe('dropTable generation-distinct stores', function () { }); it('completes a tombstoned drop by exact store name, leaving a live same-name generation alone', async function () { + await defineTable('GenTombstoneExact').dropTable(); const Old = defineTable('GenTombstoneExact', [{ name: 'blob', type: 'Blob' }]); const blob = await createBlob(Buffer.alloc(50_000, 4)); await Old.put({ id: 1, str: 'old', blob });