Skip to content

Commit e99ab38

Browse files
openclaw-clownfish[bot]TSHOGXvincentkoc
authored
fix(memory): preserve reindex rollback recovery (#92881)
* fix(memory): preserve reindex rollback recovery Co-authored-by: Shiwen Han <[email protected]> * fix(clownfish): address review for gitcrawl-5644-autonomous-smoke (1) Co-authored-by: Shiwen Han <[email protected]> * test: update memory reindex test routing expectation * chore(memory): remove release changelog entry * fix(memory): complete reindex retry recovery --------- Co-authored-by: openclaw-clownfish[bot] <280122609+openclaw-clownfish[bot]@users.noreply.github.com> Co-authored-by: Shiwen Han <[email protected]> Co-authored-by: Vincent Koc <[email protected]>
1 parent 19130e0 commit e99ab38

11 files changed

Lines changed: 726 additions & 124 deletions

‎extensions/memory-core/src/memory/embedding.test-mocks.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,8 @@ export function resetEmbeddingMocks(): void {
2828

2929
vi.mock("./embeddings.js", () => ({
3030
resolveEmbeddingProviderAdapterId: (providerId: string) => providerId,
31+
resolveEmbeddingProviderAdapterTransport: (providerId: string) =>
32+
providerId === "local" ? "local" : "remote",
3133
createEmbeddingProvider: async () => ({
3234
requestedProvider: "openai",
3335
provider: {

‎extensions/memory-core/src/memory/manager-atomic-reindex.ts‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -158,19 +158,29 @@ async function swapMemoryIndexFiles(
158158
targetPath: string,
159159
tempPath: string,
160160
options: MemoryIndexFileOptions = {},
161+
afterPublish?: () => Promise<void> | void,
161162
): Promise<void> {
162163
// On POSIX (Linux/macOS), rename(2) atomically overwrites the target,
163164
// so there is no absent-window between removing the old index and
164165
// publishing the new one. On Windows, rename fails when the target
165166
// exists, so the three-step backup protocol is retained.
166167
const resolvedOptions = resolveMemoryIndexFileOptions(options);
167168
const backupPath = `${targetPath}.backup-${randomUUID()}`;
169+
let published = false;
170+
const markPublished = async () => {
171+
if (published) {
172+
return;
173+
}
174+
published = true;
175+
await afterPublish?.();
176+
};
168177
// The old and temp DBs are checkpointed and closed before swap. Hide target
169178
// sidecars before publishing the new main DB, but keep them rollbackable
170179
// until the main-file publish succeeds.
171180
await moveMemoryIndexSidecarsWithRollback(targetPath, backupPath, resolvedOptions);
172181
try {
173182
await renameWithRetry(tempPath, targetPath, resolvedOptions);
183+
await markPublished();
174184
} catch (err) {
175185
if (
176186
(err as NodeJS.ErrnoException).code === "EPERM" ||
@@ -185,6 +195,7 @@ async function swapMemoryIndexFiles(
185195
}
186196
try {
187197
await renameWithRetry(tempPath, targetPath, resolvedOptions);
198+
await markPublished();
188199
} catch (moveErr) {
189200
await moveMemoryIndexFiles(backupPath, targetPath, options);
190201
throw moveErr;
@@ -205,11 +216,17 @@ export async function runMemoryAtomicReindex<T>(params: {
205216
tempPath: string;
206217
build: () => Promise<T>;
207218
beforeTempCleanup?: () => Promise<void> | void;
219+
afterPublish?: () => Promise<void> | void;
208220
fileOptions?: MemoryIndexFileOptions;
209221
}): Promise<T> {
210222
try {
211223
const result = await params.build();
212-
await swapMemoryIndexFiles(params.targetPath, params.tempPath, params.fileOptions);
224+
await swapMemoryIndexFiles(
225+
params.targetPath,
226+
params.tempPath,
227+
params.fileOptions,
228+
params.afterPublish,
229+
);
213230
return result;
214231
} catch (err) {
215232
try {

‎extensions/memory-core/src/memory/manager-embedding-ops.ts‎

Lines changed: 34 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -228,6 +228,36 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps {
228228
.run(excess);
229229
}
230230

231+
private upsertEmbeddingCacheEntries(
232+
entries: Array<{ hash: string; embedding: number[] }>,
233+
provider: { id: string; model: string } | null = this.provider,
234+
): void {
235+
upsertMemoryEmbeddingCache({
236+
db: this.db,
237+
enabled: this.cache.enabled,
238+
provider,
239+
providerKey: this.providerKey,
240+
entries,
241+
tableName: EMBEDDING_CACHE_TABLE,
242+
});
243+
if (this.embeddingCacheMirrorDb && this.embeddingCacheMirrorDb !== this.db) {
244+
try {
245+
upsertMemoryEmbeddingCache({
246+
db: this.embeddingCacheMirrorDb,
247+
enabled: this.cache.enabled,
248+
provider,
249+
providerKey: this.providerKey,
250+
entries,
251+
tableName: EMBEDDING_CACHE_TABLE,
252+
});
253+
} catch (err) {
254+
log.warn("memory embeddings: failed to mirror safe-reindex cache batch", {
255+
error: formatErrorMessage(err),
256+
});
257+
}
258+
}
259+
}
260+
231261
private async embedChunksInBatches(chunks: MemoryChunk[]): Promise<number[][]> {
232262
if (chunks.length === 0) {
233263
return [];
@@ -240,7 +270,6 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps {
240270

241271
const missingChunks = missing.map((m) => m.chunk);
242272
const batches = buildMemoryEmbeddingBatches(missingChunks, EMBEDDING_BATCH_MAX_TOKENS);
243-
const toCache: Array<{ hash: string; embedding: number[] }> = [];
244273
const provider = this.provider;
245274
if (!provider) {
246275
throw new Error("Cannot embed batch in FTS-only mode (no embedding provider)");
@@ -261,24 +290,18 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps {
261290
const batchEmbeddings = hasStructuredInputs
262291
? await this.embedBatchInputsWithRetry(inputs)
263292
: await this.embedBatchWithRetry(batch.map((chunk) => chunk.text));
293+
const batchCacheEntries: Array<{ hash: string; embedding: number[] }> = [];
264294
for (let i = 0; i < batch.length; i += 1) {
265295
const item = missing[cursor + i];
266296
const embedding = batchEmbeddings[i] ?? [];
267297
if (item) {
268298
embeddings[item.index] = embedding;
269-
toCache.push({ hash: item.chunk.hash, embedding });
299+
batchCacheEntries.push({ hash: item.chunk.hash, embedding });
270300
}
271301
}
302+
this.upsertEmbeddingCacheEntries(batchCacheEntries);
272303
cursor += batch.length;
273304
}
274-
upsertMemoryEmbeddingCache({
275-
db: this.db,
276-
enabled: this.cache.enabled,
277-
provider: this.provider,
278-
providerKey: this.providerKey,
279-
entries: toCache,
280-
tableName: EMBEDDING_CACHE_TABLE,
281-
});
282305
return embeddings;
283306
}
284307

@@ -354,14 +377,7 @@ export abstract class MemoryManagerEmbeddingOps extends MemoryManagerSyncOps {
354377
embeddings[item.index] = embedding;
355378
toCache.push({ hash: item.chunk.hash, embedding });
356379
}
357-
upsertMemoryEmbeddingCache({
358-
db: this.db,
359-
enabled: this.cache.enabled,
360-
provider,
361-
providerKey: this.providerKey,
362-
entries: toCache,
363-
tableName: EMBEDDING_CACHE_TABLE,
364-
});
380+
this.upsertEmbeddingCacheEntries(toCache, provider);
365381
return embeddings;
366382
}
367383

‎extensions/memory-core/src/memory/manager-session-reindex.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
export function shouldSyncSessionsForReindex(params: {
33
hasSessionSource: boolean;
44
sessionsDirty: boolean;
5+
sessionsFullRetryDirty?: boolean;
56
dirtySessionFileCount: number;
67
sync?: {
78
reason?: string;
@@ -22,6 +23,9 @@ export function shouldSyncSessionsForReindex(params: {
2223
if (params.needsFullReindex) {
2324
return true;
2425
}
26+
if (params.sessionsFullRetryDirty) {
27+
return true;
28+
}
2529
const reason = params.sync?.reason;
2630
if (reason === "session-start" || reason === "watch") {
2731
return false;

0 commit comments

Comments
 (0)