fix: import dedup off-switch honesty + brain-owned lifecycle for the background pass
The post-import background deduplication pass (a merge-DELETE writer, debounced ~5 minutes after import) had four lifecycle defects: - enableDeduplication: false did not gate the background schedule — an import that explicitly opted out could still have entities auto-removed minutes later. The flag now gates both the inline and background passes. - Each import() constructed its own coordinator-owned deduplicator, so the debounce never spanned imports (N imports = N delete timers). The brain now owns a single lazy instance (getBackgroundDeduplicator). - close() never cancelled pending dedup; a delete pass could fire against a closed brain. close() now cancels it first. - The 5-minute timer held the process open (exit-hang class); now unref'd. Four regression tests pin the contract (background-dedup-lifecycle); import guides document that false disables both passes.
This commit is contained in:
parent
120205b69c
commit
4fcef7b8ed
7 changed files with 179 additions and 10 deletions
22
RELEASES.md
22
RELEASES.md
|
|
@ -10,6 +10,28 @@ Full auto-generated changelog: `CHANGELOG.md` · Releases: https://github.com/so
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
|
## v8.8.1 — 2026-07-18 (the import dedup off-switch is now honest + lifecycle-safe)
|
||||||
|
|
||||||
|
The post-import background deduplication pass (a merge-DELETE writer that runs ~5 minutes
|
||||||
|
after an import, merging entities judged duplicates by id / name / vector similarity) had
|
||||||
|
three lifecycle defects, all fixed:
|
||||||
|
|
||||||
|
- **`enableDeduplication: false` now actually disables it.** The background pass was
|
||||||
|
scheduled unconditionally — an import that explicitly opted out could still have
|
||||||
|
entities auto-removed 5 minutes later. The flag now gates BOTH the inline merge and
|
||||||
|
the background pass (regression-pinned).
|
||||||
|
- **One deduplicator per brain, owned by the brain.** Each `import()` call constructed its
|
||||||
|
own coordinator + deduplicator, so the "debounced" timer never actually debounced across
|
||||||
|
imports (N imports = N delete timers). The brain now owns a single instance — the
|
||||||
|
debounce genuinely spans imports — and `close()` cancels pending work, so a delete pass
|
||||||
|
can never fire against a closed brain.
|
||||||
|
- **The 5-minute timer is unref'd** — a pending pass no longer holds the process open
|
||||||
|
(the exit-hang class; this timer had escaped the earlier sweep).
|
||||||
|
|
||||||
|
Retention note for keep-everything deployments: with `enableDeduplication: false` on
|
||||||
|
import calls and `retention: 'all'` in config, no engine path removes records
|
||||||
|
automatically.
|
||||||
|
|
||||||
## v8.8.0 — 2026-07-17 (OS-limit detection for pool-scale deployments)
|
## v8.8.0 — 2026-07-17 (OS-limit detection for pool-scale deployments)
|
||||||
|
|
||||||
Small minor: brains now detect the two OS limits that bite at pool scale and warn **before**
|
Small minor: brains now detect the two OS limits that bite at pool scale and warn **before**
|
||||||
|
|
|
||||||
|
|
@ -300,7 +300,10 @@ await brain.import(data, {
|
||||||
// Deduplication
|
// Deduplication
|
||||||
enableDeduplication: true, // Check for duplicate entities (default: true)
|
enableDeduplication: true, // Check for duplicate entities (default: true)
|
||||||
deduplicationThreshold: 0.85, // Similarity threshold for duplicates (0-1, default: 0.85)
|
deduplicationThreshold: 0.85, // Similarity threshold for duplicates (0-1, default: 0.85)
|
||||||
// Note: Auto-disabled for imports >100 entities
|
// Notes: false disables BOTH the inline merge and the background pass that
|
||||||
|
// runs ~5 min after the last import (merged duplicates are deleted).
|
||||||
|
// The inline pass auto-disables for imports >100 entities (O(n²) cost);
|
||||||
|
// the background pass still covers those unless the flag is false.
|
||||||
|
|
||||||
// Performance
|
// Performance
|
||||||
chunkSize: 100, // Batch size for processing (default: varies by operation)
|
chunkSize: 100, // Batch size for processing (default: varies by operation)
|
||||||
|
|
|
||||||
|
|
@ -86,11 +86,22 @@ await brain.import(file, {
|
||||||
|
|
||||||
```typescript
|
```typescript
|
||||||
await brain.import(file, {
|
await brain.import(file, {
|
||||||
enableDeduplication: true, // Check for duplicates (default: false)
|
enableDeduplication: true, // Check for duplicates (default: true)
|
||||||
deduplicationThreshold: 0.85 // Similarity threshold (default: 0.85)
|
deduplicationThreshold: 0.85 // Similarity threshold (default: 0.85)
|
||||||
})
|
})
|
||||||
```
|
```
|
||||||
|
|
||||||
|
Deduplication merges entities judged duplicates — the non-primary records are
|
||||||
|
**deleted**. Set `enableDeduplication: false` to disable it entirely: the flag
|
||||||
|
gates both the inline merge during import and the background pass that runs
|
||||||
|
about 5 minutes after the last import.
|
||||||
|
|
||||||
|
```typescript
|
||||||
|
await brain.import(file, {
|
||||||
|
enableDeduplication: false // No merging, inline or background
|
||||||
|
})
|
||||||
|
```
|
||||||
|
|
||||||
### Import Tracking
|
### Import Tracking
|
||||||
|
|
||||||
Track and organize imports by project:
|
Track and organize imports by project:
|
||||||
|
|
|
||||||
|
|
@ -10128,6 +10128,30 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
return await coordinator.import(source as Buffer | string | object, options)
|
return await coordinator.import(source as Buffer | string | object, options)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Brain-owned background deduplicator (lazy; see getBackgroundDeduplicator). */
|
||||||
|
private _backgroundDedup?: import('./import/BackgroundDeduplicator.js').BackgroundDeduplicator
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The single brain-owned BackgroundDeduplicator, lazily constructed.
|
||||||
|
*
|
||||||
|
* Ownership matters here: the post-import dedup timer must outlive the
|
||||||
|
* per-call ImportCoordinator but never the brain. One instance per brain
|
||||||
|
* restores the intended cross-import debounce (per-coordinator instances
|
||||||
|
* each armed their own timer, so the "debounce" never spanned imports) and
|
||||||
|
* gives close() a handle to cancel pending work — a delete pass must never
|
||||||
|
* fire against a closed brain.
|
||||||
|
* @internal
|
||||||
|
*/
|
||||||
|
async getBackgroundDeduplicator(): Promise<
|
||||||
|
import('./import/BackgroundDeduplicator.js').BackgroundDeduplicator
|
||||||
|
> {
|
||||||
|
if (!this._backgroundDedup) {
|
||||||
|
const { BackgroundDeduplicator } = await import('./import/BackgroundDeduplicator.js')
|
||||||
|
this._backgroundDedup = new BackgroundDeduplicator(this)
|
||||||
|
}
|
||||||
|
return this._backgroundDedup
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Virtual File System API - Knowledge Operating System
|
* Virtual File System API - Knowledge Operating System
|
||||||
*
|
*
|
||||||
|
|
@ -16070,6 +16094,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
* This ensures deferred persistence mode data is saved
|
* This ensures deferred persistence mode data is saved
|
||||||
*/
|
*/
|
||||||
async close(): Promise<void> {
|
async close(): Promise<void> {
|
||||||
|
// Cancel any pending post-import background deduplication FIRST — it is a
|
||||||
|
// writer (merge-deletes), and no delete pass may start mid- or post-close.
|
||||||
|
this._backgroundDedup?.cancelPending()
|
||||||
|
|
||||||
// Change-feed teardown: no events are delivered for or after close().
|
// Change-feed teardown: no events are delivered for or after close().
|
||||||
this._changeFeed.close()
|
this._changeFeed.close()
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -41,6 +41,14 @@ export interface DeduplicationStats {
|
||||||
* - Import-scoped deduplication (no cross-contamination)
|
* - Import-scoped deduplication (no cross-contamination)
|
||||||
* - 3-tier strategy (ID → Name → Similarity)
|
* - 3-tier strategy (ID → Name → Similarity)
|
||||||
* - Uses existing indexes (EntityIdMapper, MetadataIndexManager, TypeAware HNSW)
|
* - Uses existing indexes (EntityIdMapper, MetadataIndexManager, TypeAware HNSW)
|
||||||
|
*
|
||||||
|
* Lifecycle: ONE instance per brain, owned by Brainy (getBackgroundDeduplicator)
|
||||||
|
* so the debounce genuinely spans imports and brain.close() cancels pending
|
||||||
|
* work via cancelPending() — this pass merge-DELETES duplicate entities, so it
|
||||||
|
* must never fire against a closed brain. The enableDeduplication gate lives
|
||||||
|
* at the scheduling call site (ImportCoordinator); scheduleDedup itself is
|
||||||
|
* unconditional. The timer is unref'd — a pending pass never holds the
|
||||||
|
* process open.
|
||||||
*/
|
*/
|
||||||
export class BackgroundDeduplicator {
|
export class BackgroundDeduplicator {
|
||||||
private brain: Brainy
|
private brain: Brainy
|
||||||
|
|
@ -67,12 +75,15 @@ export class BackgroundDeduplicator {
|
||||||
clearTimeout(this.debounceTimer)
|
clearTimeout(this.debounceTimer)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Schedule for 5 minutes from now
|
// Schedule for 5 minutes from now. unref'd: a pending dedup pass must
|
||||||
|
// never hold the process open (exit-hang class) — if the process exits
|
||||||
|
// first, the pass simply never runs; imports are already durable.
|
||||||
this.debounceTimer = setTimeout(() => {
|
this.debounceTimer = setTimeout(() => {
|
||||||
this.runBatchDedup().catch(error => {
|
this.runBatchDedup().catch(error => {
|
||||||
prodLog.error('[BackgroundDedup] Batch dedup failed:', error)
|
prodLog.error('[BackgroundDedup] Batch dedup failed:', error)
|
||||||
})
|
})
|
||||||
}, 5 * 60 * 1000)
|
}, 5 * 60 * 1000)
|
||||||
|
this.debounceTimer.unref?.()
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
||||||
|
|
@ -13,7 +13,6 @@
|
||||||
import { Brainy } from '../brainy.js'
|
import { Brainy } from '../brainy.js'
|
||||||
import { FormatDetector, SupportedFormat } from './FormatDetector.js'
|
import { FormatDetector, SupportedFormat } from './FormatDetector.js'
|
||||||
import { ImportHistory, type ImportHistoryEntry } from './ImportHistory.js'
|
import { ImportHistory, type ImportHistoryEntry } from './ImportHistory.js'
|
||||||
import { BackgroundDeduplicator } from './BackgroundDeduplicator.js'
|
|
||||||
import { SmartExcelImporter } from '../importers/SmartExcelImporter.js'
|
import { SmartExcelImporter } from '../importers/SmartExcelImporter.js'
|
||||||
import { SmartPDFImporter } from '../importers/SmartPDFImporter.js'
|
import { SmartPDFImporter } from '../importers/SmartPDFImporter.js'
|
||||||
import { SmartCSVImporter } from '../importers/SmartCSVImporter.js'
|
import { SmartCSVImporter } from '../importers/SmartCSVImporter.js'
|
||||||
|
|
@ -112,7 +111,12 @@ export interface ValidImportOptions {
|
||||||
/** Confidence threshold for entities */
|
/** Confidence threshold for entities */
|
||||||
confidenceThreshold?: number
|
confidenceThreshold?: number
|
||||||
|
|
||||||
/** Enable entity deduplication across imports */
|
/**
|
||||||
|
* Enable entity deduplication (default: true). Gates BOTH passes: the
|
||||||
|
* inline merge during import AND the debounced background pass that runs
|
||||||
|
* ~5 minutes after the last import (which merge-DELETES duplicate entities).
|
||||||
|
* Set false for deployments that must never auto-remove records.
|
||||||
|
*/
|
||||||
enableDeduplication?: boolean
|
enableDeduplication?: boolean
|
||||||
|
|
||||||
/** Similarity threshold for deduplication (0-1) */
|
/** Similarity threshold for deduplication (0-1) */
|
||||||
|
|
@ -286,7 +290,6 @@ export class ImportCoordinator {
|
||||||
private brain: Brainy
|
private brain: Brainy
|
||||||
private detector: FormatDetector
|
private detector: FormatDetector
|
||||||
private history: ImportHistory
|
private history: ImportHistory
|
||||||
private backgroundDedup: BackgroundDeduplicator
|
|
||||||
private excelImporter: SmartExcelImporter
|
private excelImporter: SmartExcelImporter
|
||||||
private pdfImporter: SmartPDFImporter
|
private pdfImporter: SmartPDFImporter
|
||||||
private csvImporter: SmartCSVImporter
|
private csvImporter: SmartCSVImporter
|
||||||
|
|
@ -300,7 +303,6 @@ export class ImportCoordinator {
|
||||||
this.brain = brain
|
this.brain = brain
|
||||||
this.detector = new FormatDetector()
|
this.detector = new FormatDetector()
|
||||||
this.history = new ImportHistory(brain)
|
this.history = new ImportHistory(brain)
|
||||||
this.backgroundDedup = new BackgroundDeduplicator(brain)
|
|
||||||
this.excelImporter = new SmartExcelImporter(brain)
|
this.excelImporter = new SmartExcelImporter(brain)
|
||||||
this.pdfImporter = new SmartPDFImporter(brain)
|
this.pdfImporter = new SmartPDFImporter(brain)
|
||||||
this.csvImporter = new SmartCSVImporter(brain)
|
this.csvImporter = new SmartCSVImporter(brain)
|
||||||
|
|
@ -1459,9 +1461,16 @@ export class ImportCoordinator {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Schedule background deduplication (debounced 5 minutes)
|
// Schedule background deduplication (debounced 5 minutes, brain-owned so
|
||||||
if (trackingContext && trackingContext.importId) {
|
// close() can cancel it). Honors the same enableDeduplication gate as the
|
||||||
this.backgroundDedup.scheduleDedup(trackingContext.importId)
|
// inline pass — false means NO dedup, inline or background.
|
||||||
|
if (
|
||||||
|
trackingContext &&
|
||||||
|
trackingContext.importId &&
|
||||||
|
options.enableDeduplication !== false
|
||||||
|
) {
|
||||||
|
const backgroundDedup = await this.brain.getBackgroundDeduplicator()
|
||||||
|
backgroundDedup.scheduleDedup(trackingContext.importId)
|
||||||
}
|
}
|
||||||
|
|
||||||
return {
|
return {
|
||||||
|
|
|
||||||
85
tests/integration/background-dedup-lifecycle.test.ts
Normal file
85
tests/integration/background-dedup-lifecycle.test.ts
Normal file
|
|
@ -0,0 +1,85 @@
|
||||||
|
/**
|
||||||
|
* @module tests/integration/background-dedup-lifecycle
|
||||||
|
* @description The post-import background deduplication pass (a merge-DELETE
|
||||||
|
* writer) obeys the same contract as the inline pass. Laws:
|
||||||
|
* (1) enableDeduplication:false schedules NO background pass — the brain-owned
|
||||||
|
* deduplicator is never even constructed;
|
||||||
|
* (2) by default the pass IS scheduled, brain-owned, with an unref'd timer
|
||||||
|
* (a pending pass never holds the process open);
|
||||||
|
* (3) repeated imports debounce into ONE pending batch on ONE instance
|
||||||
|
* (per-coordinator instances used to arm one timer per import);
|
||||||
|
* (4) close() cancels pending work — no delete pass can fire after close.
|
||||||
|
*/
|
||||||
|
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
|
||||||
|
import { Brainy } from '../../src/brainy.js'
|
||||||
|
|
||||||
|
const ROWS = [
|
||||||
|
{ name: 'Alice Zephyr', role: 'engineer' },
|
||||||
|
{ name: 'Bob Quill', role: 'writer' }
|
||||||
|
]
|
||||||
|
|
||||||
|
// Keep imports fast and deterministic — dedup scheduling is what's under test.
|
||||||
|
const FAST = {
|
||||||
|
enableNeuralExtraction: false,
|
||||||
|
enableRelationshipInference: false,
|
||||||
|
enableConceptExtraction: false
|
||||||
|
} as const
|
||||||
|
|
||||||
|
// Deterministic stub embedder (hnsw-rebuild.test.ts pattern) — dedup
|
||||||
|
// scheduling never inspects vector CONTENT, so skip the WASM model load.
|
||||||
|
const stubEmbedding = async (text: string): Promise<number[]> => {
|
||||||
|
const hash = text.split('').reduce((acc, char) => acc + char.charCodeAt(0), 0)
|
||||||
|
const vector = new Array(384).fill(0).map((_, i) => Math.sin(hash + i))
|
||||||
|
return vector
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('background dedup lifecycle', () => {
|
||||||
|
let brain: Brainy
|
||||||
|
|
||||||
|
beforeEach(async () => {
|
||||||
|
brain = new Brainy({
|
||||||
|
requireSubtype: false,
|
||||||
|
storage: { type: 'memory' as const },
|
||||||
|
embeddingFunction: stubEmbedding
|
||||||
|
})
|
||||||
|
await brain.init()
|
||||||
|
})
|
||||||
|
|
||||||
|
afterEach(async () => {
|
||||||
|
await brain.close()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('enableDeduplication:false schedules no background pass at all', async () => {
|
||||||
|
await brain.import(ROWS, { ...FAST, enableDeduplication: false })
|
||||||
|
expect((brain as any)._backgroundDedup).toBeUndefined()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('default schedules a brain-owned pass with an unref-ed timer', async () => {
|
||||||
|
await brain.import(ROWS, { ...FAST })
|
||||||
|
const dedup = (brain as any)._backgroundDedup
|
||||||
|
expect(dedup).toBeDefined()
|
||||||
|
expect(dedup.pendingImports.size).toBe(1)
|
||||||
|
const timer = dedup.debounceTimer
|
||||||
|
expect(timer).toBeDefined()
|
||||||
|
// Node timers expose hasRef(); an unref'd timer must not hold the process.
|
||||||
|
expect(typeof timer.hasRef).toBe('function')
|
||||||
|
expect(timer.hasRef()).toBe(false)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('imports debounce into one pending batch on one brain-owned instance', async () => {
|
||||||
|
await brain.import(ROWS, { ...FAST })
|
||||||
|
const first = (brain as any)._backgroundDedup
|
||||||
|
await brain.import([{ name: 'Cara Vex', role: 'analyst' }], { ...FAST })
|
||||||
|
expect((brain as any)._backgroundDedup).toBe(first)
|
||||||
|
expect(first.pendingImports.size).toBe(2)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('close() cancels pending background dedup', async () => {
|
||||||
|
await brain.import(ROWS, { ...FAST })
|
||||||
|
const dedup = (brain as any)._backgroundDedup
|
||||||
|
expect(dedup.debounceTimer).toBeDefined()
|
||||||
|
await brain.close()
|
||||||
|
expect(dedup.debounceTimer).toBeUndefined()
|
||||||
|
expect(dedup.pendingImports.size).toBe(0)
|
||||||
|
})
|
||||||
|
})
|
||||||
Loading…
Add table
Add a link
Reference in a new issue