watch: слежение за файлами (WatchService + debounce + реконсиляция)
This commit is contained in:
@@ -0,0 +1,90 @@
|
||||
package memo.watch
|
||||
|
||||
import memo.core.Db
|
||||
import memo.core.Embedder
|
||||
import memo.core.Indexer
|
||||
import java.io.File
|
||||
import java.util.concurrent.CountDownLatch
|
||||
import kotlin.system.exitProcess
|
||||
|
||||
fun main(args: Array<String>) {
|
||||
if (args.isEmpty()) {
|
||||
System.err.println("usage: watch <path>")
|
||||
exitProcess(2)
|
||||
}
|
||||
val raw = File(args[0])
|
||||
val base = if (raw.name == ".memo") raw.parentFile ?: raw else raw
|
||||
val collections = discoverCollections(base)
|
||||
if (collections.isEmpty()) {
|
||||
System.err.println("коллекции не найдены в ${base.absolutePath}")
|
||||
exitProcess(1)
|
||||
}
|
||||
|
||||
val modelDir = System.getenv("MEMO_MODEL_DIR") ?: "/root/WORK/memo/models/siglip2"
|
||||
val modelPath = "$modelDir/text_model_int8.onnx"
|
||||
val tokenizerPath = "$modelDir/tokenizer.model"
|
||||
|
||||
val ctxList = collections.map { coll ->
|
||||
val memoDir = File(coll, ".memo")
|
||||
memoDir.mkdirs()
|
||||
val db = Db(File(memoDir, "index.db").absolutePath)
|
||||
db.init()
|
||||
val embedder = Embedder(modelPath, tokenizerPath)
|
||||
val indexer = Indexer(db, embedder)
|
||||
CollCtx(coll, db, embedder, indexer)
|
||||
}
|
||||
|
||||
val indexersByCollection = ctxList.associateBy { it.root }
|
||||
val watcher = Watcher(
|
||||
collections = ctxList.map { it.root },
|
||||
index = { coll -> indexersByCollection.getValue(coll).indexer.indexTree(coll) },
|
||||
)
|
||||
|
||||
Runtime.getRuntime().addShutdownHook(
|
||||
Thread({
|
||||
try { watcher.close() } catch (_: Throwable) {}
|
||||
for (ctx in ctxList) {
|
||||
try { ctx.embedder.close() } catch (_: Throwable) {}
|
||||
try { ctx.db.close() } catch (_: Throwable) {}
|
||||
}
|
||||
}, "memo-watch-shutdown")
|
||||
)
|
||||
|
||||
watcher.start()
|
||||
for (ctx in ctxList) {
|
||||
println("наблюдаю: ${ctx.root.absolutePath}")
|
||||
}
|
||||
|
||||
CountDownLatch(1).await()
|
||||
}
|
||||
|
||||
private data class CollCtx(
|
||||
val root: File,
|
||||
val db: Db,
|
||||
val embedder: Embedder,
|
||||
val indexer: Indexer,
|
||||
)
|
||||
|
||||
private fun discoverCollections(base: File): List<File> {
|
||||
if (!base.isDirectory) return emptyList()
|
||||
val candidates = LinkedHashSet<File>()
|
||||
candidates.add(base)
|
||||
val q = ArrayDeque<Pair<File, Int>>()
|
||||
q.addLast(base to 0)
|
||||
while (q.isNotEmpty()) {
|
||||
val (d, depth) = q.removeFirst()
|
||||
if (depth >= 2) continue
|
||||
val children = d.listFiles() ?: continue
|
||||
for (c in children) {
|
||||
if (c.isDirectory && !c.name.startsWith(".")) {
|
||||
candidates.add(c)
|
||||
q.addLast(c to depth + 1)
|
||||
}
|
||||
}
|
||||
}
|
||||
return candidates.filter { d ->
|
||||
d.walkTopDown()
|
||||
.maxDepth(8)
|
||||
.any { it.isFile && it.extension == "md" }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,141 @@
|
||||
package memo.watch
|
||||
|
||||
import java.io.File
|
||||
import java.nio.file.ClosedWatchServiceException
|
||||
import java.nio.file.FileSystems
|
||||
import java.nio.file.Path
|
||||
import java.nio.file.StandardWatchEventKinds
|
||||
import java.nio.file.WatchKey
|
||||
import java.nio.file.WatchService
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
import java.util.concurrent.TimeUnit
|
||||
import java.util.concurrent.atomic.AtomicBoolean
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
|
||||
class Watcher(
|
||||
private val collections: List<File>,
|
||||
private val index: (File) -> Unit,
|
||||
private val debounceMillis: Long = 500,
|
||||
private val reconcileMillis: Long = 10 * 60 * 1000,
|
||||
) : AutoCloseable {
|
||||
|
||||
private val watchService: WatchService = FileSystems.getDefault().newWatchService()
|
||||
private val keyToCollection = ConcurrentHashMap<WatchKey, File>()
|
||||
private val running = AtomicBoolean(false)
|
||||
private val closed = AtomicBoolean(false)
|
||||
private val _indexCalls = AtomicInteger(0)
|
||||
|
||||
val indexCalls: Int get() = _indexCalls.get()
|
||||
|
||||
private val thread: Thread = Thread({ runLoop() }, "memo-watcher").apply { isDaemon = true }
|
||||
|
||||
fun start() {
|
||||
if (!running.compareAndSet(false, true)) return
|
||||
for (coll in collections) {
|
||||
try {
|
||||
registerRecursive(coll, coll)
|
||||
} catch (t: Throwable) {
|
||||
System.err.println("watcher: register ${coll.absolutePath} failed: ${t.message}")
|
||||
}
|
||||
}
|
||||
thread.start()
|
||||
}
|
||||
|
||||
private fun registerRecursive(dir: File, collection: File) {
|
||||
if (!dir.isDirectory) return
|
||||
if (dir.name.startsWith(".")) return
|
||||
val key = dir.toPath().register(
|
||||
watchService,
|
||||
StandardWatchEventKinds.ENTRY_CREATE,
|
||||
StandardWatchEventKinds.ENTRY_MODIFY,
|
||||
StandardWatchEventKinds.ENTRY_DELETE,
|
||||
)
|
||||
keyToCollection[key] = collection
|
||||
val children = dir.listFiles() ?: return
|
||||
for (child in children) {
|
||||
if (child.isDirectory && !child.name.startsWith(".")) {
|
||||
registerRecursive(child, collection)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override fun close() {
|
||||
if (!closed.compareAndSet(false, true)) return
|
||||
running.set(false)
|
||||
thread.interrupt()
|
||||
try { watchService.close() } catch (_: Throwable) {}
|
||||
try { thread.join(2000) } catch (_: InterruptedException) {}
|
||||
}
|
||||
|
||||
private fun runLoop() {
|
||||
var nextReconcile = System.currentTimeMillis() + reconcileMillis
|
||||
while (running.get()) {
|
||||
try {
|
||||
val firstKey = watchService.poll(debounceMillis, TimeUnit.MILLISECONDS)
|
||||
if (firstKey != null) {
|
||||
val affected = HashSet<File>()
|
||||
processKey(firstKey, affected)
|
||||
val endDeadline = System.currentTimeMillis() + debounceMillis
|
||||
while (running.get()) {
|
||||
val now = System.currentTimeMillis()
|
||||
if (now >= endDeadline) break
|
||||
val remaining = endDeadline - now
|
||||
val nextKey = watchService.poll(remaining, TimeUnit.MILLISECONDS) ?: break
|
||||
processKey(nextKey, affected)
|
||||
}
|
||||
for (coll in affected) {
|
||||
callIndex(coll)
|
||||
}
|
||||
}
|
||||
if (System.currentTimeMillis() >= nextReconcile) {
|
||||
nextReconcile = System.currentTimeMillis() + reconcileMillis
|
||||
for (coll in collections) {
|
||||
callIndex(coll)
|
||||
}
|
||||
}
|
||||
} catch (_: ClosedWatchServiceException) {
|
||||
break
|
||||
} catch (_: InterruptedException) {
|
||||
if (!running.get()) break
|
||||
} catch (t: Throwable) {
|
||||
System.err.println("watcher loop: ${t.message}")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun processKey(key: WatchKey, affected: MutableSet<File>) {
|
||||
val collection = keyToCollection[key] ?: return
|
||||
for (event in key.pollEvents()) {
|
||||
val kind = event.kind()
|
||||
if (kind === StandardWatchEventKinds.OVERFLOW) continue
|
||||
val ctx = event.context() as? Path ?: continue
|
||||
val watchable = key.watchable() as? Path ?: continue
|
||||
val ev = watchable.resolve(ctx).toFile()
|
||||
when (kind) {
|
||||
StandardWatchEventKinds.ENTRY_CREATE -> {
|
||||
affected.add(collection)
|
||||
if (ev.isDirectory && !ev.name.startsWith(".")) {
|
||||
try {
|
||||
registerRecursive(ev, collection)
|
||||
} catch (t: Throwable) {
|
||||
System.err.println("watcher: registerRecursive ${ev.absolutePath} failed: ${t.message}")
|
||||
}
|
||||
}
|
||||
}
|
||||
StandardWatchEventKinds.ENTRY_MODIFY,
|
||||
StandardWatchEventKinds.ENTRY_DELETE -> affected.add(collection)
|
||||
}
|
||||
}
|
||||
val valid = key.reset()
|
||||
if (!valid) keyToCollection.remove(key)
|
||||
}
|
||||
|
||||
private fun callIndex(coll: File) {
|
||||
_indexCalls.incrementAndGet()
|
||||
try {
|
||||
index(coll)
|
||||
} catch (t: Throwable) {
|
||||
System.err.println("watcher: index ${coll.absolutePath} failed: ${t.message}")
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
package memo.watch
|
||||
|
||||
import java.io.File
|
||||
import java.nio.file.Files
|
||||
import java.util.concurrent.TimeUnit
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
class WatcherTest {
|
||||
|
||||
@Test
|
||||
fun modifyTriggersSingleIndexCall() {
|
||||
val tempDir = Files.createTempDirectory("memo-watch-").toFile()
|
||||
tempDir.deleteOnExit()
|
||||
val note = File(tempDir, "note.md")
|
||||
note.writeText("# initial")
|
||||
note.deleteOnExit()
|
||||
val counter = AtomicInteger(0)
|
||||
val watcher = Watcher(
|
||||
collections = listOf(tempDir),
|
||||
index = { counter.incrementAndGet() },
|
||||
debounceMillis = 200,
|
||||
reconcileMillis = TimeUnit.HOURS.toMillis(1),
|
||||
)
|
||||
try {
|
||||
watcher.start()
|
||||
Thread.sleep(300)
|
||||
note.appendText("\nappended")
|
||||
assertTrue(
|
||||
waitFor(5_000, 100) { counter.get() >= 1 },
|
||||
"index должен сработать в течение 5с, получено ${counter.get()}",
|
||||
)
|
||||
Thread.sleep(1500)
|
||||
assertEquals(1, counter.get(), "ожидался ровно 1 вызов index, получено ${counter.get()}")
|
||||
} finally {
|
||||
watcher.close()
|
||||
tempDir.deleteRecursively()
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun unchangedFileDoesNotTriggerIndex() {
|
||||
val tempDir = Files.createTempDirectory("memo-watch-").toFile()
|
||||
tempDir.deleteOnExit()
|
||||
val note = File(tempDir, "note.md")
|
||||
note.writeText("# initial")
|
||||
note.deleteOnExit()
|
||||
val counter = AtomicInteger(0)
|
||||
val watcher = Watcher(
|
||||
collections = listOf(tempDir),
|
||||
index = { counter.incrementAndGet() },
|
||||
debounceMillis = 200,
|
||||
reconcileMillis = TimeUnit.HOURS.toMillis(1),
|
||||
)
|
||||
try {
|
||||
watcher.start()
|
||||
Thread.sleep(300)
|
||||
Thread.sleep(1500)
|
||||
assertEquals(0, counter.get(), "не должно быть вызовов index, получено ${counter.get()}")
|
||||
} finally {
|
||||
watcher.close()
|
||||
tempDir.deleteRecursively()
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun closeIsIdempotentAndStopsThread() {
|
||||
val tempDir = Files.createTempDirectory("memo-watch-").toFile()
|
||||
tempDir.deleteOnExit()
|
||||
val watcher = Watcher(
|
||||
collections = listOf(tempDir),
|
||||
index = {},
|
||||
)
|
||||
watcher.start()
|
||||
watcher.close()
|
||||
watcher.close()
|
||||
tempDir.deleteRecursively()
|
||||
}
|
||||
|
||||
private fun waitFor(timeoutMs: Long, stepMs: Long, condition: () -> Boolean): Boolean {
|
||||
val deadline = System.currentTimeMillis() + timeoutMs
|
||||
while (System.currentTimeMillis() < deadline) {
|
||||
if (condition()) return true
|
||||
Thread.sleep(stepMs)
|
||||
}
|
||||
return condition()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user