Skip to content

fix(plugin): close dirty-closed rescan gaps: cursor starvation, cancelled-drain settlement, mid-drain re-emission #2588

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
syzsunshine219 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix(plugin): close dirty-closed rescan gaps: cursor starvation, cancelled-drain settlement, mid-drain re-emission #2588

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1ebc0db
fix(plugin): close dirty-closed rescan gaps: cursor starvation and ca…
akushonkamen Oct 12, 2026
c70019c
fix(plugin): stop periodic dirty rescan from re-emitting episodes mid…
akushonkamen Oct 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
149 changes: 116 additions & 33 deletions apps/memos-local-plugin/core/pipeline/memory-core.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,10 @@ import {
FLOAT32_BYTES,
} from "../storage/repos/index.js";
import type { EmbeddingCountsBucket } from "../storage/repos/index.js";
import type { ClosedEpisodeCursor } from "../storage/repos/episodes.js";
import type {
ClosedEpisodeCursor,
EpisodeMetaRow,
} from "../storage/repos/episodes.js";
import { createEmbedder } from "../embedding/embedder.js";
import { createLlmClient } from "../llm/client.js";
import {
Expand Down Expand Up @@ -727,6 +730,14 @@ export function createMemoryCore(
let startupRecoveryCancelled = false;
let lastStaleScan = 0;
let lastDirtyClosedScan = 0;
// Issue #2459 (review): episode ids whose dirty-rescore drain is still
// in flight (phase-1 emitted, phase-2 not yet settled). The head-page
// pass re-reads the newest page on every 10-minute tick, and an
// in-flight episode still carries `rewardDirty` until settlement —
// without this guard the periodic scan re-collected and re-emitted the
// whole batch every tick (the duplicate LLM spend gap 2 set out to
// eliminate). Cleared per episode as soon as its settlement runs.
const dirtyRescoreInFlight = new Set<EpisodeId>();
async function autoFinalizeStaleTasks(): Promise<void> {
if (!autoRecoveryEnabled) return;
const nowMs = Date.now();
Expand Down Expand Up @@ -800,7 +811,14 @@ export function createMemoryCore(
if (handle.algorithm.lightweightMemory.enabled) return;
try {
const allDirty = collectDirtyClosedEpisodes()
.filter((ep) => !isLightweightEpisode(ep) && episodeRewardIsDirty(ep));
.filter((ep) => !isLightweightEpisode(ep) && episodeRewardIsDirty(ep))
// Issue #2459 (review): rows whose drain is still mid-flight
// keep `rewardDirty` until phase-2 settles — skip them here so a
// tick racing an open drain cannot re-emit and re-run the whole
// batch (duplicate LLM spend). Freshly dirtied rows are NOT
// blocked by this: they are not in flight, so they are still
// collected on the very same tick.
.filter((ep) => !dirtyRescoreInFlight.has(ep.id as EpisodeId));
// Apply the same backoff filter as init() so the 10-min periodic
// scan does not hammer episodes whose LLM call keeps failing.
const dirtyClosed = allDirty.filter((ep) => dirtyEpisodeBackoffElapsed(ep, nowMs));
Expand Down Expand Up @@ -1197,7 +1215,12 @@ export function createMemoryCore(
} else {
const nowForDirty = Date.now();
const allDirty = collectDirtyClosedEpisodes()
.filter((ep) => !isLightweightEpisode(ep) && episodeRewardIsDirty(ep));
.filter((ep) => !isLightweightEpisode(ep) && episodeRewardIsDirty(ep))
// Issue #2459 (review): same in-flight guard as the periodic
// scan — rows already mid-drain keep `rewardDirty` until
// settlement, and re-emitting them here would queue duplicate
// reward runs.
.filter((ep) => !dirtyRescoreInFlight.has(ep.id as EpisodeId));
const dirtyClosed: typeof allDirty = [];
for (const ep of allDirty) {
if (dirtyEpisodeBackoffElapsed(ep, nowForDirty)) {
Expand All @@ -1216,7 +1239,11 @@ export function createMemoryCore(
dirtyClosedForBackground = dirtyClosed;
}
} catch (err) {
log.debug("init.orphan_scan.failed", {
// Issue #2459: a skipped startup classification used to be
// invisible at default log levels — the reporter only found it
// by diffing kv cursor writes. This failure means zero
// collection for the whole boot, so it must be visible.
log.warn("init.orphan_scan.failed", {
err: err instanceof Error ? err.message : String(err),
});
}
Expand Down Expand Up @@ -1679,6 +1706,9 @@ export function createMemoryCore(
? prevDirty.failedAttempts
: 0;
priorFailedAttempts.set(episodeId, prevAttempts);
// Mark in-flight BEFORE emitting so a periodic tick racing this
// drain cannot queue a duplicate run for the same row.
dirtyRescoreInFlight.add(episodeId);
handle.repos.episodes.updateMeta(episodeId, {
closeReason: "finalized",
recoveredAtStartup: endedAt,
Expand All @@ -1693,31 +1723,58 @@ export function createMemoryCore(
closedBy: "finalized",
});
}
await handle.flush();
if (startupRecoveryCancelled) return;
// After the reward / reflect chain has finished, account for the
// outcome: clear `meta.rewardDirty` on episodes that are no longer
// dirty (success), bump `failedAttempts + lastFailureAt` on episodes
// that still match the dirty predicate (LLM failure / no-op). This
// closes the "retried indefinitely" loop reported in issue #1808.
const now = Date.now();
for (const [episodeId, prevAttempts] of priorFailedAttempts) {
const after = handle.repos.episodes.getById(episodeId);
if (!after) continue;
const stillDirty = episodeRewardIsDirty(after);
if (stillDirty) {
handle.repos.episodes.updateMeta(episodeId, {
rewardDirty: {
failedAttempts: prevAttempts + 1,
lastFailureAt: now,
},
});
} else if (
after.meta &&
typeof after.meta === "object" &&
"rewardDirty" in after.meta
) {
handle.repos.episodes.updateMeta(episodeId, { rewardDirty: undefined });
try {
await handle.flush();
} finally {
// After the reward / reflect chain has finished, account for the
// outcome: clear `meta.rewardDirty` on episodes that are no longer
// dirty (success), bump `failedAttempts + lastFailureAt` on
// episodes that still match the dirty predicate (LLM failure /
// no-op). This closes the "retried indefinitely" loop reported in
// issue #1808.
//
// Issue #2459 (gap 2): a shutdown cancel mid-drain used to `return`
// right after the flush, skipping this settlement for the *whole*
// batch — episodes whose chains had already rescored successfully
// kept their `rewardDirty` markers and were re-collected + re-run
// later (duplicate LLM spend), while failures lost their backoff
// counters. Settling is synchronous bookkeeping on rows whose
// chains already ran, so it happens even when cancelled (and even
// if the flush itself threw mid-teardown) — best-effort, one
// try/catch per row so a single late-teardown error cannot drop
// the rest of the batch.
const now = Date.now();
for (const [episodeId, prevAttempts] of priorFailedAttempts) {
try {
const after = handle.repos.episodes.getById(episodeId);
if (after) {
const stillDirty = episodeRewardIsDirty(after);
if (stillDirty) {
handle.repos.episodes.updateMeta(episodeId, {
rewardDirty: {
failedAttempts: prevAttempts + 1,
lastFailureAt: now,
},
});
} else if (
after.meta &&
typeof after.meta === "object" &&
"rewardDirty" in after.meta
) {
handle.repos.episodes.updateMeta(episodeId, {
rewardDirty: undefined,
});
}
}
} catch (err) {
log.warn("init.dirty_closed_episodes.settle_failed", {
episodeId,
err: err instanceof Error ? err.message : String(err),
});
}
// The chain for this row has reached a terminal state (settled
// or un-settleable) — let the next scan consider it again.
dirtyRescoreInFlight.delete(episodeId);
}
}
}
Expand All @@ -1728,10 +1785,36 @@ export function createMemoryCore(
if (storedCursor !== null && !cursor) {
handle.repos.kv.del(DIRTY_CLOSED_SCAN_CURSOR_KEY);
}
const page = handle.repos.episodes.listClosedPage({
// Issue #2459 (gap 1): the stored cursor only ever moves *away* from
// the newest rows, so an episode that becomes dirty after the cursor
// paged past its position was not revisited until the walk reached
// the end of the closed set and the cursor reset — a full lap of
// `ceil(closed / 500) × 10 min` (≈70 min on the reporter's DB,
// hours on the 30k+-episode DBs of #1787). Take one bounded page
// from the newest end on every scan: freshly closed episodes and
// freshly re-marked recent rows are seen immediately, while the
// cursor walk below keeps draining the older backlog a page at a
// time.
const collected = new Map<string, EpisodeRow & EpisodeMetaRow>();
const headPage = handle.repos.episodes.listClosedPage({
limit: DIRTY_CLOSED_SCAN_PAGE_SIZE,
before: cursor,
before: null,
});
for (const ep of headPage) {
collected.set(ep.id, ep);
}
// Without a stored cursor the head page *is* the walk's first page;
// only page again when a cursor is in play.
let page = headPage;
if (cursor) {
page = handle.repos.episodes.listClosedPage({
limit: DIRTY_CLOSED_SCAN_PAGE_SIZE,
before: cursor,
});
for (const ep of page) {
collected.set(ep.id, ep);
}
}
if (page.length === 0) {
handle.repos.kv.del(DIRTY_CLOSED_SCAN_CURSOR_KEY);
return [];
Expand All @@ -1750,7 +1833,7 @@ export function createMemoryCore(
// `rewardDirty` marker on short/trivial episodes. The skip is terminal;
// clean the contradictory marker while this bounded page is already in
// memory so upgraded installations stop retrying those rows immediately.
for (const ep of page) {
for (const ep of collected.values()) {
if (!rewardWasSkipped(ep) || !hasRewardDirtyMarker(ep)) continue;
try {
handle.repos.episodes.updateMeta(ep.id as EpisodeId, {
Expand All @@ -1767,7 +1850,7 @@ export function createMemoryCore(
});
}
}
return page;
return [...collected.values()];
}

function isDirtyClosedScanCursor(value: unknown): value is ClosedEpisodeCursor {
Expand Down
Loading
Loading