diff --git a/server/services/brainSyncLog.js b/server/services/brainSyncLog.js index e3a00ed9f..ff384654c 100644 --- a/server/services/brainSyncLog.js +++ b/server/services/brainSyncLog.js @@ -237,9 +237,38 @@ export async function getChangesSince(sinceSeq, limit = 100) { } /** - * Compact the log by dropping entries below minSeq + * Replay entries for a single (type, id) according to runtime LWW rules, + * returning the surviving terminal entry. + * + * Incumbent wins ties (same timestamp) matching applyRemoteRecord. + * Entries missing updatedAt are skipped. + */ +function replayTerminal(entries) { + let accepted = null; + for (const e of [...entries].sort((a, b) => a.seq - b.seq)) { + const ts = e?.record?.updatedAt; + if (ts == null) continue; + if (accepted == null || ts > accepted.record.updatedAt) accepted = e; + } + return accepted; +} + +/** + * Compact the sync log using a compatibility-preserving compaction representation. + * + * Deltas at or above `minSeq` (unconsumed by at least one active peer) are + * preserved verbatim in sequence order. + * + * For history below `minSeq` (or all entries when minSeq is 0, e.g. on installs + * with no brain-sync peers), redundant intermediate updates are pruned by + * retaining only the surviving terminal LWW entry per (type, id). + * + * The durable sequence counter (maxSeq) is always determined from disk state + * under the log mutex, preventing index skew after a failed append, and is + * preserved so initSyncLog recovers the monotonic sequence counter across + * restarts. */ -export async function compactLog(minSeq) { +export async function compactLog(minSeq = 0) { return withLock(async () => { await ensureBrainDir(); // Load first: the rebuild below marks the index loaded, so skipping this @@ -248,33 +277,126 @@ export async function compactLog(minSeq) { if (!existsSync(SYNC_LOG_FILE)) return 0; const content = await readFile(SYNC_LOG_FILE, 'utf-8'); - const lines = content.trim().split('\n').filter(l => l.trim()); - const kept = []; - let dropped = 0; - - for (const line of lines) { - const entry = safeJSONParse(line, null); - if (!entry || entry.seq < minSeq) { - dropped++; - continue; + const rawLines = content.trim().split('\n').filter(l => l.trim()); + if (rawLines.length === 0) return 0; + + const parsedLines = []; + let maxDurableSeq = 0; + let maxSeqEntry = null; + + for (const rawLine of rawLines) { + const entry = safeJSONParse(rawLine, null); + if (entry && typeof entry.seq === 'number') { + if (entry.seq > maxDurableSeq) { + maxDurableSeq = entry.seq; + maxSeqEntry = entry; + } + parsedLines.push({ rawLine, entry, seq: entry.seq }); + } else { + // Line without numeric seq (e.g. malformed or unindexed note) + parsedLines.push({ rawLine, entry: null, seq: null }); + } + } + + const floor = typeof minSeq === 'number' && Number.isFinite(minSeq) + ? Math.max(0, Math.min(minSeq, maxDurableSeq)) + : 0; + + const preservedTail = []; + const tailKeys = new Set(); + const olderEntriesByKey = new Map(); + const unindexedOrUntypedOlder = []; + + for (const item of parsedLines) { + const { entry, seq } = item; + if (seq !== null && floor > 0 && seq >= floor) { + preservedTail.push(item); + if (entry?.type && entry?.id) { + tailKeys.add(`${entry.type}/${entry.id}`); + } + } else if (entry?.type && entry?.id) { + const key = `${entry.type}/${entry.id}`; + if (!olderEntriesByKey.has(key)) olderEntriesByKey.set(key, []); + olderEntriesByKey.get(key).push(item); + } else { + if (floor === 0 || seq === null) { + unindexedOrUntypedOlder.push(item); + } + } + } + + // Replay terminal winning state for older keys + const keptOlder = []; + const olderWinnersByKey = new Map(); + for (const [key, items] of olderEntriesByKey) { + const entries = items.map(i => i.entry); + const olderWinner = replayTerminal(entries); + if (!olderWinner) continue; + + // If this key also appears in the preserved tail, check whether any tail + // operation strictly supersedes the pre-floor LWW winner (updatedAt > olderWinner.updatedAt). + // If the tail carries ONLY stale/losing operations (e.g. olderWinner is a Jan-02 delete + // and tail has an echoed Jan-01 create), we MUST retain olderWinner before the verbatim + // tail so fresh / delta-only peers do not accept the stale create and resurrect the record. + if (tailKeys.has(key)) { + const tailItems = preservedTail.filter(i => i.entry && `${i.entry.type}/${i.entry.id}` === key); + const supersededByTail = tailItems.some(i => { + const tailTs = i.entry?.record?.updatedAt; + return tailTs != null && olderWinner.record?.updatedAt != null && tailTs > olderWinner.record.updatedAt; + }); + if (supersededByTail) { + continue; + } + } + + const matchingItem = items.find(i => i.entry === olderWinner) + || { rawLine: JSON.stringify(olderWinner), entry: olderWinner, seq: olderWinner.seq }; + keptOlder.push(matchingItem); + olderWinnersByKey.set(key, matchingItem); + } + + const kept = [...unindexedOrUntypedOlder, ...keptOlder, ...preservedTail]; + + // Ensure the durable max sequence is preserved so restart recovery and cursors hold + if (maxSeqEntry && !kept.some(i => i.seq === maxSeqEntry.seq)) { + if (!maxSeqEntry.type || !maxSeqEntry.id) { + kept.push({ rawLine: JSON.stringify(maxSeqEntry), entry: maxSeqEntry, seq: maxSeqEntry.seq }); + } else { + const key = `${maxSeqEntry.type}/${maxSeqEntry.id}`; + const winner = olderWinnersByKey.get(key); + if (winner && winner.entry) { + winner.entry.seq = maxSeqEntry.seq; + winner.seq = maxSeqEntry.seq; + winner.rawLine = JSON.stringify(winner.entry); + } else { + kept.push({ rawLine: JSON.stringify(maxSeqEntry), entry: maxSeqEntry, seq: maxSeqEntry.seq }); + } } - kept.push({ line, seq: entry.seq }); } - const newContent = kept.length > 0 ? kept.map(k => k.line).join('\n') + '\n' : ''; + // Sort kept entries: items with numeric seq sorted by seq + kept.sort((a, b) => { + if (a.seq !== null && b.seq !== null) return a.seq - b.seq; + return 0; + }); + + const dropped = rawLines.length - kept.length; + if (dropped <= 0) return 0; + + const newContent = kept.map(i => i.rawLine).join('\n') + '\n'; await atomicWrite(SYNC_LOG_FILE, newContent); - // Rebuild the index from what we just wrote — the offsets all moved. A kept - // line without a numeric seq keeps its bytes but stays OUT of the index: - // a non-numeric seq in the array would break firstIndexAfter's binary - // search and hide every entry before it from peers. + // Rebuild index offsets from what was written offsets = []; let offset = 0; - for (const { line, seq } of kept) { - if (typeof seq === 'number') offsets.push({ seq, offset }); - offset += Buffer.byteLength(line, 'utf8') + 1; + for (const { rawLine, seq } of kept) { + if (typeof seq === 'number') { + offsets.push({ seq, offset }); + } + offset += Buffer.byteLength(rawLine, 'utf8') + 1; } fileSize = offset; + currentSeq = maxDurableSeq; pendingNewline = false; indexLoaded = true; diff --git a/server/services/brainSyncLog.test.js b/server/services/brainSyncLog.test.js index 20df7df6c..8a65d9c8a 100644 --- a/server/services/brainSyncLog.test.js +++ b/server/services/brainSyncLog.test.js @@ -20,7 +20,11 @@ function getTempRoot() { vi.mock('../lib/fileUtils.js', async () => { const actual = await vi.importActual('../lib/fileUtils.js'); - return makePathsProxy(actual, { dataRoot: () => getTempRoot() }); + const proxy = makePathsProxy(actual, { dataRoot: () => getTempRoot() }); + return { + ...proxy, + atomicWrite: vi.fn(actual.atomicWrite) + }; }); // Spy on createReadStream so the tests can assert WHICH bytes were read — that @@ -39,6 +43,7 @@ vi.mock('fs/promises', async () => { import { createReadStream } from 'fs'; import { appendFile } from 'fs/promises'; +import { atomicWrite } from '../lib/fileUtils.js'; import { initSyncLog, getCurrentSeq, @@ -392,9 +397,11 @@ describe('brainSyncLog', () => { }); it('serves reads from the rebuilt index immediately after compaction', async () => { - for (let i = 0; i < 5; i++) { - await appendChange('create', 'people', `p${i}`, { name: `Person ${i}` }, 'inst-1'); - } + await appendChange('create', 'people', 'p0', { name: 'Person 0 v1' }, 'inst-1'); + await appendChange('update', 'people', 'p0', { name: 'Person 0 v2' }, 'inst-1'); + await appendChange('update', 'people', 'p0', { name: 'Person 0 v3' }, 'inst-1'); + await appendChange('create', 'people', 'p1', { name: 'Person 1' }, 'inst-1'); + await appendChange('create', 'people', 'p2', { name: 'Person 2' }, 'inst-1'); expect(await compactLog(3)).toBe(2); @@ -423,16 +430,142 @@ describe('brainSyncLog', () => { }); it('keeps appending at the right offset after compaction', async () => { - for (let i = 0; i < 4; i++) { - await appendChange('create', 'people', `p${i}`, { name: `Person ${i}` }, 'inst-1'); - } + await appendChange('create', 'people', 'p0', { name: 'Person 0 v1' }, 'inst-1'); + await appendChange('update', 'people', 'p0', { name: 'Person 0 v2' }, 'inst-1'); + await appendChange('create', 'people', 'p1', { name: 'Person 1' }, 'inst-1'); + await appendChange('create', 'people', 'p2', { name: 'Person 2' }, 'inst-1'); await compactLog(3); - await appendChange('create', 'people', 'p5', { name: 'Person 5' }, 'inst-1'); + await appendChange('create', 'people', 'p3', { name: 'Person 3' }, 'inst-1'); vi.clearAllMocks(); const result = await getChangesSince(4); expect(lastReadStart()).toBe(lineOffset(2)); expect(result.changes.map(c => c.seq)).toEqual([5]); }); + + it('prunes intermediate update churn to terminal survivors and preserves max seq on installs with no peers (#5439)', async () => { + await appendChange('create', 'people', 'p0', { name: 'Person 0 v1', updatedAt: '2026-01-01T00:00:00.000Z' }, 'inst-1'); + await appendChange('update', 'people', 'p0', { name: 'Person 0 v2', updatedAt: '2026-01-02T00:00:00.000Z' }, 'inst-1'); + await appendChange('update', 'people', 'p0', { name: 'Person 0 v3', updatedAt: '2026-01-03T00:00:00.000Z' }, 'inst-1'); + await appendChange('create', 'people', 'p1', { name: 'Person 1', updatedAt: '2026-01-01T00:00:00.000Z' }, 'inst-1'); + await appendChange('delete', 'people', 'p1', { updatedAt: '2026-01-04T00:00:00.000Z' }, 'inst-1'); + await appendChange('create', 'people', 'p2', { name: 'Person 2', updatedAt: '2026-01-05T00:00:00.000Z' }, 'inst-1'); + expect(getCurrentSeq()).toBe(6); + + // Compact with floor 0 (compatibility-preserving compaction representation) + const dropped = await compactLog(0); + expect(dropped).toBe(3); // 2 intermediate p0 updates + 1 p1 create superseded by delete + + // Verify file on disk contains the 3 terminal survivors + const lines = readFileSync(syncLogPath(), 'utf8').trim().split('\n'); + expect(lines).toHaveLength(3); + const parsed = lines.map(l => JSON.parse(l)); + expect(parsed.map(p => ({ id: p.id, op: p.op }))).toEqual([ + { id: 'p0', op: 'update' }, + { id: 'p1', op: 'delete' }, + { id: 'p2', op: 'create' } + ]); + expect(parsed.map(p => p.seq)).toEqual([3, 5, 6]); + + // All active entities are available for delta pulls (e.g. pre-#1077 / fresh peers) + const deltaResult = await getChangesSince(0); + expect(deltaResult.changes).toHaveLength(3); + + // Simulate restart: re-initialize the log + await initSyncLog(); + expect(getCurrentSeq()).toBe(6); + + // Subsequent appends continue monotonic sequence numbers + const nextEntry = await appendChange('create', 'people', 'p3', { name: 'Person 3' }, 'inst-1'); + expect(nextEntry.seq).toBe(7); + expect(getCurrentSeq()).toBe(7); + }); + + it('returns 0 and does not call atomicWrite when no entries are dropped (#5439)', async () => { + for (let i = 0; i < 3; i++) { + await appendChange('create', 'people', `p${i}`, { name: `Person ${i}`, updatedAt: `2026-01-0${i + 1}T00:00:00.000Z` }, 'inst-1'); + } + atomicWrite.mockClear(); + + // Calling compactLog with minSeq <= 1 drops nothing + const dropped = await compactLog(1); + expect(dropped).toBe(0); + expect(atomicWrite).not.toHaveBeenCalled(); + }); + + it('determines compaction floor strictly under mutex from durable state after a failed append', async () => { + for (let i = 0; i < 3; i++) { + await appendChange('create', 'people', `p${i}`, { name: `Person ${i}`, updatedAt: `2026-01-0${i + 1}T00:00:00.000Z` }, 'inst-1'); + } + expect(getCurrentSeq()).toBe(3); + + // Simulate appendFile disk failure during appendChange + appendFile.mockRejectedValueOnce(new Error('Disk I/O error')); + await expect(appendChange('create', 'people', 'p3', { name: 'Person 3' }, 'inst-1')).rejects.toThrow('Disk I/O error'); + + // In-memory currentSeq was reserved/incremented to 4, but disk holds up to seq 3 + expect(getCurrentSeq()).toBe(4); + + // Compacting under mutex re-syncs durable state from disk and does NOT wipe the log + const dropped = await compactLog(4); + expect(dropped).toBe(0); + + // All 3 durable entries on disk remain intact + const lines = readFileSync(syncLogPath(), 'utf8').trim().split('\n'); + expect(lines).toHaveLength(3); + expect(getCurrentSeq()).toBe(3); + + // Sequence recovery across restart recovers the durable seq 3 + await initSyncLog(); + expect(getCurrentSeq()).toBe(3); + const next = await appendChange('create', 'people', 'p3', { name: 'Person 3' }, 'inst-1'); + expect(next.seq).toBe(4); + }); + + it('replays LWW rules correctly in terminal compaction', async () => { + writeLog( + '{"seq":1,"op":"create","type":"links","id":"x","record":{"updatedAt":"2026-01-01T00:00:00.000Z"}}\n' + + '{"seq":2,"op":"delete","type":"links","id":"x","record":{"updatedAt":"2026-01-02T00:00:00.000Z"}}\n' + + '{"seq":99,"op":"create","type":"links","id":"x","record":{"updatedAt":"2026-01-01T00:00:00.000Z"}}\n' + ); + await initSyncLog(); + expect(await compactLog(0)).toBe(2); + + const lines = readFileSync(syncLogPath(), 'utf8').trim().split('\n').map(l => JSON.parse(l)); + expect(lines).toHaveLength(1); + expect(lines[0].op).toBe('delete'); // LWW winner + expect(lines[0].record.updatedAt).toBe('2026-01-02T00:00:00.000Z'); + expect(lines[0].seq).toBe(99); // max seq preserved + }); + + it('preserves pre-floor LWW winner before verbatim tail when tail is stale under positive floor', async () => { + // seq 1: create links x (Jan-01) + // seq 2: delete links x (Jan-02) -> winning delete + // seq 99: stale create links x (Jan-01) -> stale echoed create in tail + writeLog( + '{"seq":1,"op":"create","type":"links","id":"x","record":{"updatedAt":"2026-01-01T00:00:00.000Z"}}\n' + + '{"seq":2,"op":"delete","type":"links","id":"x","record":{"updatedAt":"2026-01-02T00:00:00.000Z"}}\n' + + '{"seq":99,"op":"create","type":"links","id":"x","record":{"updatedAt":"2026-01-01T00:00:00.000Z"}}\n' + ); + await initSyncLog(); + + // Compact with positive floor (50). Tail (seq >= 50) is kept verbatim, + // and seq 2 (delete Jan-02) is preserved before the tail so a fresh peer + // replaying from since=0 rejects the stale create and avoids resurrecting the record. + const dropped = await compactLog(50); + expect(dropped).toBe(1); // seq 1 is dropped, seq 2 and seq 99 are kept + + const lines = readFileSync(syncLogPath(), 'utf8').trim().split('\n').map(l => JSON.parse(l)); + expect(lines).toHaveLength(2); + expect(lines[0].seq).toBe(2); + expect(lines[0].op).toBe('delete'); + expect(lines[0].record.updatedAt).toBe('2026-01-02T00:00:00.000Z'); + expect(lines[1].seq).toBe(99); + expect(lines[1].op).toBe('create'); + + // A fresh peer pulling from since=0 gets seq 2 (delete) then seq 99 (stale create) + const fromZero = await getChangesSince(0); + expect(fromZero.changes.map(c => c.seq)).toEqual([2, 99]); + }); }); }); diff --git a/server/services/syncOrchestrator.js b/server/services/syncOrchestrator.js index 5c6f2dcf1..d746f49d1 100644 --- a/server/services/syncOrchestrator.js +++ b/server/services/syncOrchestrator.js @@ -867,14 +867,19 @@ export async function syncAllPeers() { // runs (vs. the old skip-when-empty) so a 0 floor is an explicit "keep all", // and the anti-entropy reconcile (Part 1) re-converges anyone genuinely behind. const brainPeers = peers.filter(p => p.enabled && p.instanceId && getEffectiveCategories(p).brain); + let minSeq = 0; if (brainPeers.length > 0) { const consumedSeqs = brainPeers.map(p => { const consumed = p.remoteSyncSeqs?.cursorForYou?.brainSeq; return typeof consumed === 'number' && Number.isFinite(consumed) && consumed >= 0 ? consumed : 0; }); - const minSeq = Math.min(...consumedSeqs); - await brainSyncLog.compactLog(minSeq); + minSeq = Math.min(...consumedSeqs); } + // When brain peers exist, minSeq preserves unconsumed deltas above the floor. + // When no brain peers are enabled (or on standalone installs, #5439), floor 0 + // runs compatibility-preserving compaction: pruning intermediate update churn + // while retaining terminal state for inbound, asymmetric, or pre-#1077 consumers. + await brainSyncLog.compactLog(minSeq); } /** diff --git a/server/services/syncOrchestrator.test.js b/server/services/syncOrchestrator.test.js index 3cce6b4c1..642479040 100644 --- a/server/services/syncOrchestrator.test.js +++ b/server/services/syncOrchestrator.test.js @@ -847,6 +847,29 @@ describe('syncOrchestrator', () => { expect(compactLog).toHaveBeenCalledWith(0); }); + + it('calls compactLog(0) for compatibility-preserving compaction when peers are configured but none is brain-enabled (#5439)', async () => { + const memoryOnlyPeer = { + ...mockPeer, + instanceId: 'A', + syncCategories: { brain: false, memory: true } + }; + getPeers.mockResolvedValue([memoryOnlyPeer]); + readJSONFile.mockImplementation(async () => ({})); + + await syncAllPeers(); + + expect(compactLog).toHaveBeenCalledWith(0); + }); + + it('calls compactLog(0) for compatibility-preserving compaction when no peers are configured (#5439)', async () => { + getPeers.mockResolvedValue([]); + readJSONFile.mockImplementation(async () => ({})); + + await syncAllPeers(); + + expect(compactLog).toHaveBeenCalledWith(0); + }); }); describe('initSyncOrchestrator', () => {