Repository navigation
Expand file tree
/
Copy pathpath_cache.js
More file actions
228 lines (204 loc) · 8.51 KB
/
Copy pathpath_cache.js
File metadata and controls
228 lines (204 loc) · 8.51 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
const { open } = require("lmdb");
const fs = require("fs-extra");
const path = require("path");
class PathCache {
constructor(config, logger) {
this.config = config;
this.logger = logger;
this.lmdb = null;
this.dbs = {};
}
async init(modelConfigs) {
if (!this.config.lmdb) return;
this.logger.info("📦 Initializing LMDB...");
try {
// Only open a database for models that are actually enabled.
// LMDB caps the number of open named databases (maxDbs); opening one per entry in
// the config file made that cap depend on the SIZE OF THE CONFIG rather than on
// what the orchestrator actually uses. A consumer listing 61 models but enabling 22
// hit MDB_DBS_FULL at startup, which leaves the cache empty and silently breaks
// every lookup. Models without an explicit `enabled` flag stay enabled (backwards
// compatible with configs predating the flag).
const modelKeys = Object.keys(modelConfigs.models).filter(
(key) => modelConfigs.models[key].enabled !== false
);
this.lmdb = open({
path: this.config.lmdb.path,
compression: true,
mapSize: this.config.lmdb.mapSize || 104857600,
// Headroom above what we open, overridable by the consumer.
maxDbs: this.config.lmdb.maxDbs || Math.max(60, modelKeys.length * 2)
});
// Open named databases for each enabled model
this.dbs = {};
for (const modelKey of modelKeys) {
this.dbs[modelKey] = this.lmdb.openDB({
name: modelKey
});
}
this.logger.info(`✅ LMDB initialized (${modelKeys.length} model DBs)`);
} catch (err) {
this.logger.error("❌ Failed to initialize LMDB:", err);
throw err;
}
}
async close() {
if (this.lmdb) {
await this.lmdb.close();
this.logger.info("db LMDB closed");
this.lmdb = null;
this.dbs = {};
}
}
get(modelKey, timestamp) {
if (!this.lmdb || !this.dbs[modelKey]) return null;
return this.dbs[modelKey].get(timestamp);
}
async set(modelKey, timestamp, fullPath) {
if (!this.lmdb || !this.dbs[modelKey]) return;
const db = this.dbs[modelKey];
try {
// 1. Get old path
const oldPath = db.get(timestamp);
// 2. Put new path
await db.put(timestamp, fullPath);
this.logger.debug(`LMDB updated for model: ${modelKey}, key: ${timestamp}`);
// 3. Delete old file if different and exists
if (oldPath && oldPath !== fullPath) {
try {
if(await fs.pathExists(oldPath)) {
await fs.remove(oldPath);
this.logger.info(`🗑️ Deleted old version: ${oldPath}`);
}
} catch (delError) {
this.logger.warn(`⚠️ Failed to delete old file ${oldPath}: ${delError.message}`);
}
}
return true;
} catch (err) {
this.logger.error(`❌ LMDB update failed: ${err.message}`);
return false;
}
}
/**
* Retire la clé `timestamp` SI elle désigne encore `filePath` — appelé juste après la
* suppression du fichier, pour que l'index ne survive jamais à ce qu'il référence.
* La garde sur le chemin évite d'effacer une clé repointée sur un run plus récent.
*/
async dropKeyForFile(modelKey, timestamp, filePath) {
const db = this.dbs[modelKey];
if (!db) return false;
// Comparaison sur les chemins RÉSOLUS, pas sur les chaînes : les producteurs
// n'écrivent pas tous la même forme (l'orchestrateur joint sur `cacheDir`, une
// ingestion externe — modèle `managed: false` — peut indexer en absolu). Avec un
// `cacheDir` relatif les deux formes ne se rencontraient jamais, donc la garde
// refusait toujours : le fichier partait, la clé restait et désignait un absent.
const stored = db.get(timestamp);
if (!stored || path.resolve(stored) !== path.resolve(filePath)) return false;
await db.remove(timestamp);
return true;
}
/**
* Supprime les clés dont l'ÉCHÉANCE est plus vieille que `retentionHours`.
* (La purge des fichiers, elle, se base sur leur mtime — voir cleanupModelFiles.)
*/
async cleanup(modelKey, modelName, retentionHours) {
if (!this.lmdb || !this.dbs[modelKey]) return 0;
const db = this.dbs[modelKey];
// Les clés sont des `YYYY-MM-DDTHHMM` de largeur fixe : leur ordre lexicographique
// est chronologique, donc le cutoff formaté pareil borne le curseur nativement.
const cutoff = new Date(Date.now() - retentionHours * 60 * 60 * 1000).toISOString();
const cutoffKey = cutoff.slice(0, 13) + cutoff.slice(14, 16);
try {
const keysToDelete = [...db.getKeys({ end: cutoffKey })];
if (keysToDelete.length === 0) {
this.logger.debug(`🗑️ No old LMDB keys to delete for ${modelKey}.`);
return 0;
}
this.logger.debug(`🗑️ Deleting ${keysToDelete.length} old LMDB keys for ${modelKey}...`);
// Sans `await` par clé, lmdb-js regroupe le tout dans une seule transaction.
let last;
for (const key of keysToDelete) last = db.remove(key);
await last;
this.logger.info(`🧹 Cleaned ${keysToDelete.length} LMDB keys for ${modelName}`);
return keysToDelete.length;
} catch (err) {
this.logger.error(`❌ Error iterating/deleting keys for ${modelKey}:`, err.message);
return 0;
}
}
async rebuild(modelConfigs, fakeMode = false) {
if (!this.lmdb) return;
this.logger.info("🏗️ Starting LMDB index rebuild...");
let totalKeysRestored = 0;
for (const [modelKey, model] of Object.entries(modelConfigs.models)) {
// `init()` n'ouvre une base que pour les modèles activés : sans elle, scanAndIndex
// parcourrait l'arborescence pour jeter le résultat.
if (!this.dbs[modelKey]) continue;
const modelDir = path.join(this.config.storage.cacheDir, modelKey);
if (await fs.pathExists(modelDir)) {
try {
const count = await this.scanAndIndex(modelDir, modelKey, fakeMode);
if (count > 0) {
this.logger.info(`🏗️ Restored ${count} keys for ${model.name}`);
totalKeysRestored += count;
}
} catch (err) {
this.logger.error(`❌ Error rebuilding index for ${model.name}:`, err.message);
}
}
}
const orphans = await this.pruneMissingFiles();
this.logger.info(
`✅ LMDB index rebuild completed. Total keys restored: ${totalKeysRestored}`
+ (orphans ? `, ${orphans} orphan keys dropped` : "")
);
}
/**
* Retire les clés dont le fichier n'existe plus. Sans ça, un fichier purgé sur sa mtime
* laisse une entrée d'index que le consommateur résout en chemin mort.
*/
async pruneMissingFiles() {
let dropped = 0;
for (const [modelKey, db] of Object.entries(this.dbs)) {
const morts = [];
for (const { key, value } of db.getRange()) {
if (!(await fs.pathExists(value))) morts.push(key);
}
let last;
for (const key of morts) last = db.remove(key);
if (last) await last;
if (morts.length) {
this.logger.info(`🧹 Dropped ${morts.length} orphan LMDB keys for ${modelKey}`);
dropped += morts.length;
}
}
return dropped;
}
async scanAndIndex(dirPath, modelKey, fakeMode) {
let count = 0;
// withFileTypes : le type d'entrée est gratuit, un fs.stat par fichier ne l'est pas
// (et l'ancienneté ne sert pas ici, contrairement à la purge).
const items = await fs.readdir(dirPath, { withFileTypes: true });
const db = this.dbs[modelKey];
let last;
for (const item of items) {
const fullPath = path.join(dirPath, item.name);
if (item.isDirectory()) {
count += await this.scanAndIndex(fullPath, modelKey, fakeMode);
continue;
}
const isDataFile = item.name.endsWith('.om') || (fakeMode && item.name.endsWith('.txt'));
if (!isDataFile || !db) continue;
// Clé = nom de fichier sans extension (ex. 2025-12-14T2100)
const timestamp = item.name.replace(path.extname(item.name), '');
if (!db.get(timestamp)) {
last = db.put(timestamp, fullPath); // groupé en une transaction, cf. cleanup
count++;
}
}
if (last) await last;
return count;
}
}
module.exports = PathCache;