mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-21 09:32:24 +00:00
🔄 fix: Reindex MeiliSearch Documents After Indexed-Projection Changes (#15484)
* 🔄 fix: Reindex MeiliSearch Documents After Indexed-Projection Changes - Introduce _meiliIndexSchemaVersion schema field and MEILI_INDEX_SCHEMA_VERSION constant - Stamp schema version on single-document and batch index writes - Query unindexed and stale-version documents in syncWithMeili - Count stale documents in pendingIndexing in getSyncProgress - Add spec coverage for reindexing stale-version documents Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * 🔄 test: use the registered Meili conversation model Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * 🔄 style: satisfy Prettier in Meili indexing progress counts --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: Danny Avila <danny@librechat.ai>
This commit is contained in:
csdivad
GitHub
parent
ce704f5c11
commit
8a40e9f675
2 changed files with 108 additions and 24 deletions
|
|
@ -1,7 +1,8 @@
|
|||
import mongoose from 'mongoose';
|
||||
import { EModelEndpoint } from 'librechat-data-provider';
|
||||
import { MongoMemoryServer } from 'mongodb-memory-server';
|
||||
import mongoMeili, { type SchemaWithMeiliMethods } from '~/models/plugins/mongoMeili';
|
||||
import type { SchemaWithMeiliMethods } from '~/models/plugins/mongoMeili';
|
||||
import mongoMeili, { MEILI_INDEX_SCHEMA_VERSION } from '~/models/plugins/mongoMeili';
|
||||
import { createConversationModel } from '~/models/convo';
|
||||
import { createMessageModel } from '~/models/message';
|
||||
|
||||
|
|
@ -13,6 +14,7 @@ interface DynamicMeiliDocument extends mongoose.Document {
|
|||
expiredAt?: Date | null;
|
||||
_meiliIndex?: boolean;
|
||||
_meiliIndexAttempted?: boolean;
|
||||
_meiliIndexSchemaVersion?: number;
|
||||
_meiliCleanupVersion?: number;
|
||||
}
|
||||
|
||||
|
|
@ -1317,6 +1319,7 @@ describe('Meilisearch Mongoose plugin', () => {
|
|||
title: 'Indexed',
|
||||
endpoint: EModelEndpoint.openAI,
|
||||
_meiliIndex: true,
|
||||
_meiliIndexSchemaVersion: MEILI_INDEX_SCHEMA_VERSION,
|
||||
expiredAt: null,
|
||||
});
|
||||
|
||||
|
|
@ -1336,6 +1339,36 @@ describe('Meilisearch Mongoose plugin', () => {
|
|||
expect(progress.isComplete).toBe(false);
|
||||
});
|
||||
|
||||
test('reindexes documents from an older indexed schema version', async () => {
|
||||
const conversationModel = mongoose.models.Conversation as SchemaWithMeiliMethods;
|
||||
await conversationModel.deleteMany({});
|
||||
|
||||
await conversationModel.collection.insertOne({
|
||||
conversationId: new mongoose.Types.ObjectId().toString(),
|
||||
user: new mongoose.Types.ObjectId(),
|
||||
title: 'Legacy Indexed Conversation',
|
||||
endpoint: EModelEndpoint.openAI,
|
||||
_meiliIndex: true,
|
||||
_meiliIndexSchemaVersion: MEILI_INDEX_SCHEMA_VERSION - 1,
|
||||
expiredAt: null,
|
||||
});
|
||||
|
||||
const progress = await conversationModel.getSyncProgress();
|
||||
|
||||
expect(progress.totalProcessed).toBe(0);
|
||||
expect(progress.pendingIndexing).toBe(1);
|
||||
|
||||
await conversationModel.syncWithMeili();
|
||||
|
||||
expect(
|
||||
(
|
||||
await conversationModel
|
||||
.findOne({ title: 'Legacy Indexed Conversation' })
|
||||
.select('+_meiliIndexSchemaVersion')
|
||||
)?._meiliIndexSchemaVersion,
|
||||
).toBe(MEILI_INDEX_SCHEMA_VERSION);
|
||||
});
|
||||
|
||||
test('getSyncProgress excludes TTL documents from counts', async () => {
|
||||
const conversationModel = createConversationModel(
|
||||
mongoose,
|
||||
|
|
@ -1349,6 +1382,7 @@ describe('Meilisearch Mongoose plugin', () => {
|
|||
title: 'Syncable Indexed',
|
||||
endpoint: EModelEndpoint.openAI,
|
||||
_meiliIndex: true,
|
||||
_meiliIndexSchemaVersion: MEILI_INDEX_SCHEMA_VERSION,
|
||||
expiredAt: null,
|
||||
});
|
||||
|
||||
|
|
@ -1399,6 +1433,7 @@ describe('Meilisearch Mongoose plugin', () => {
|
|||
user: new mongoose.Types.ObjectId(),
|
||||
isCreatedByUser: true,
|
||||
_meiliIndex: true,
|
||||
_meiliIndexSchemaVersion: MEILI_INDEX_SCHEMA_VERSION,
|
||||
expiredAt: null,
|
||||
});
|
||||
|
||||
|
|
@ -1408,6 +1443,7 @@ describe('Meilisearch Mongoose plugin', () => {
|
|||
user: new mongoose.Types.ObjectId(),
|
||||
isCreatedByUser: false,
|
||||
_meiliIndex: true,
|
||||
_meiliIndexSchemaVersion: MEILI_INDEX_SCHEMA_VERSION,
|
||||
expiredAt: null,
|
||||
});
|
||||
|
||||
|
|
@ -2158,6 +2194,7 @@ describe('Meilisearch Mongoose plugin', () => {
|
|||
endpoint: EModelEndpoint.openAI,
|
||||
expiredAt: null,
|
||||
_meiliIndex: true,
|
||||
_meiliIndexSchemaVersion: MEILI_INDEX_SCHEMA_VERSION,
|
||||
},
|
||||
]);
|
||||
|
||||
|
|
@ -2196,6 +2233,7 @@ describe('Meilisearch Mongoose plugin', () => {
|
|||
isCreatedByUser: true,
|
||||
expiredAt: null,
|
||||
_meiliIndex: true,
|
||||
_meiliIndexSchemaVersion: MEILI_INDEX_SCHEMA_VERSION,
|
||||
},
|
||||
{
|
||||
messageId: new mongoose.Types.ObjectId(),
|
||||
|
|
|
|||
|
|
@ -32,6 +32,7 @@ interface MeiliIndexable {
|
|||
_meiliIndex?: boolean;
|
||||
_meiliIndexAttempted?: boolean;
|
||||
_meiliIndexVersion?: string;
|
||||
_meiliIndexSchemaVersion?: number;
|
||||
_meiliCleanupVersion?: number;
|
||||
}
|
||||
|
||||
|
|
@ -48,6 +49,7 @@ interface _DocumentWithMeiliIndex extends Document {
|
|||
_meiliIndex?: boolean;
|
||||
_meiliIndexAttempted?: boolean;
|
||||
_meiliIndexVersion?: string;
|
||||
_meiliIndexSchemaVersion?: number;
|
||||
_meiliCleanupVersion?: number;
|
||||
isTemporary?: boolean;
|
||||
expiredAt?: Date | null;
|
||||
|
|
@ -107,6 +109,9 @@ const getSyncConfig = () => ({
|
|||
const hasSchemaPath = (schema: Schema, path: string): boolean =>
|
||||
Object.prototype.hasOwnProperty.call(schema.obj, path);
|
||||
|
||||
/** Bump when the indexed document shape or projection changes. */
|
||||
export const MEILI_INDEX_SCHEMA_VERSION = 1;
|
||||
|
||||
const explicitTemporaryFlagKey = 'meiliExplicitTemporaryFlag';
|
||||
const previouslyIndexedFlagKey = 'meiliPreviouslyIndexed';
|
||||
const meiliCleanupVersion = 1;
|
||||
|
|
@ -304,6 +309,7 @@ const createMeiliMongooseModel = ({
|
|||
'+_meiliIndex',
|
||||
'+_meiliIndexAttempted',
|
||||
'+_meiliIndexVersion',
|
||||
'+_meiliIndexSchemaVersion',
|
||||
'isTemporary',
|
||||
'expiredAt',
|
||||
'unfinished',
|
||||
|
|
@ -378,7 +384,13 @@ const createMeiliMongooseModel = ({
|
|||
// eslint-disable-next-line no-restricted-syntax -- versioned internal bookkeeping must not re-enter document middleware
|
||||
const acknowledgement = await doc.collection.updateOne(
|
||||
{ _id: doc._id as Types.ObjectId, _meiliIndexVersion: version },
|
||||
{ $set: { _meiliIndex: true, _meiliCleanupVersion: meiliCleanupVersion } },
|
||||
{
|
||||
$set: {
|
||||
_meiliIndex: true,
|
||||
_meiliCleanupVersion: meiliCleanupVersion,
|
||||
_meiliIndexSchemaVersion: MEILI_INDEX_SCHEMA_VERSION,
|
||||
},
|
||||
},
|
||||
);
|
||||
if (acknowledgement.matchedCount > 0) {
|
||||
return;
|
||||
|
|
@ -412,36 +424,49 @@ const createMeiliMongooseModel = ({
|
|||
static async getSyncProgress(this: SchemaWithMeiliMethods): Promise<SyncProgress> {
|
||||
const indexableQuery = getIndexableQuery();
|
||||
const excludedIndexedQuery = getExcludedIndexedQuery();
|
||||
const [totalDocuments, indexedDocuments, pendingIndexing, pendingCleanup] = await Promise.all(
|
||||
[
|
||||
this.countDocuments(indexableQuery),
|
||||
this.countDocuments({
|
||||
...indexableQuery,
|
||||
_meiliIndex: true,
|
||||
}),
|
||||
this.countDocuments({
|
||||
...indexableQuery,
|
||||
_meiliIndex: { $ne: true },
|
||||
_meiliIndexAttempted: true,
|
||||
}),
|
||||
excludedIndexedQuery == null
|
||||
? Promise.resolve(0)
|
||||
: this.countDocuments(excludedIndexedQuery),
|
||||
const needsIndexingQuery: FilterQuery<unknown> = {
|
||||
$and: [
|
||||
indexableQuery,
|
||||
{
|
||||
$or: [
|
||||
{ _meiliIndex: { $ne: true }, _meiliIndexAttempted: true },
|
||||
{
|
||||
_meiliIndex: true,
|
||||
_meiliIndexSchemaVersion: { $ne: MEILI_INDEX_SCHEMA_VERSION },
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
);
|
||||
};
|
||||
const [totalDocuments, totalProcessed, pendingIndexing, pendingCleanup] = await Promise.all([
|
||||
this.countDocuments(indexableQuery),
|
||||
this.countDocuments({
|
||||
$and: [
|
||||
indexableQuery,
|
||||
{
|
||||
_meiliIndex: true,
|
||||
_meiliIndexSchemaVersion: MEILI_INDEX_SCHEMA_VERSION,
|
||||
},
|
||||
],
|
||||
}),
|
||||
this.countDocuments(needsIndexingQuery),
|
||||
excludedIndexedQuery == null
|
||||
? Promise.resolve(0)
|
||||
: this.countDocuments(excludedIndexedQuery),
|
||||
]);
|
||||
|
||||
return {
|
||||
totalProcessed: indexedDocuments,
|
||||
totalProcessed,
|
||||
totalDocuments,
|
||||
pendingIndexing,
|
||||
pendingCleanup,
|
||||
isComplete: indexedDocuments === totalDocuments && pendingCleanup === 0,
|
||||
isComplete: totalProcessed === totalDocuments && pendingCleanup === 0,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Synchronizes data between the MongoDB collection and the MeiliSearch index by
|
||||
* incrementally indexing only non-temporary documents where `_meiliIndex` is not `true`.
|
||||
* incrementally indexing only non-temporary documents that are unindexed or stale.
|
||||
* */
|
||||
static async syncWithMeili(this: SchemaWithMeiliMethods): Promise<void> {
|
||||
const startTime = Date.now();
|
||||
|
|
@ -474,8 +499,18 @@ const createMeiliMongooseModel = ({
|
|||
while (hasMore) {
|
||||
const indexableQuery = getIndexableQuery();
|
||||
const query: FilterQuery<unknown> = {
|
||||
...indexableQuery,
|
||||
_meiliIndex: { $ne: true },
|
||||
$and: [
|
||||
indexableQuery,
|
||||
{
|
||||
$or: [
|
||||
{ _meiliIndex: { $ne: true } },
|
||||
{
|
||||
_meiliIndex: true,
|
||||
_meiliIndexSchemaVersion: { $ne: MEILI_INDEX_SCHEMA_VERSION },
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
try {
|
||||
|
|
@ -548,7 +583,13 @@ const createMeiliMongooseModel = ({
|
|||
// original conversation/message timestamps (fixes sidebar chronological sort).
|
||||
await this.updateMany(
|
||||
{ _id: { $in: docsIds } },
|
||||
{ $set: { _meiliIndex: true, _meiliCleanupVersion: meiliCleanupVersion } },
|
||||
{
|
||||
$set: {
|
||||
_meiliIndex: true,
|
||||
_meiliCleanupVersion: meiliCleanupVersion,
|
||||
_meiliIndexSchemaVersion: MEILI_INDEX_SCHEMA_VERSION,
|
||||
},
|
||||
},
|
||||
{ timestamps: false },
|
||||
);
|
||||
} catch (error) {
|
||||
|
|
@ -912,6 +953,11 @@ export default function mongoMeili(schema: Schema, options: MongoMeiliOptions):
|
|||
required: false,
|
||||
select: false,
|
||||
},
|
||||
_meiliIndexSchemaVersion: {
|
||||
type: Number,
|
||||
required: false,
|
||||
select: false,
|
||||
},
|
||||
_meiliCleanupVersion: {
|
||||
type: Number,
|
||||
required: false,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue