diff --git a/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt b/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt index 05854a63c..ac3d5734c 100644 --- a/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt +++ b/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsEventStore.kt @@ -79,10 +79,12 @@ class FsEventStore( private val slots: FsSlots private val tombstones: FsTombstones private val planner: FsQueryPlanner + private val lockManager: FsLockManager init { layout.ensureSkeleton() cleanStaging() + lockManager = FsLockManager(root) hasher = TagNameValueHasher(layout.readOrCreateSeed()) indexer = FsIndexer(layout, hasher, indexingStrategy) slots = FsSlots(layout, indexer) @@ -94,7 +96,12 @@ class FsEventStore( // Insert // ------------------------------------------------------------------ - override fun insert(event: Event) { + override fun insert(event: Event) = + lockManager.withWriteLock { + insertLocked(event) + } + + private fun insertLocked(event: Event) { if (event.kind.isEphemeral()) return if (isAlreadyExpired(event)) return if (isBlockedByTombstone(event)) return @@ -231,13 +238,14 @@ class FsEventStore( } } - override fun transaction(body: IEventStore.ITransaction.() -> Unit) { - val txn = - object : IEventStore.ITransaction { - override fun insert(event: Event) = this@FsEventStore.insert(event) - } - txn.body() - } + override fun transaction(body: IEventStore.ITransaction.() -> Unit) = + lockManager.withWriteLock { + val txn = + object : IEventStore.ITransaction { + override fun insert(event: Event) = insertLocked(event) + } + txn.body() + } // ------------------------------------------------------------------ // Query @@ -304,20 +312,27 @@ class FsEventStore( // Delete // ------------------------------------------------------------------ - override fun delete(filter: Filter) { - val ids = ArrayList() - query(filter) { ids.add(it.id) } - ids.forEach { delete(it) } - } + override fun delete(filter: Filter) = + lockManager.withWriteLock { + val ids = ArrayList() + query(filter) { ids.add(it.id) } + ids.forEach { deleteLocked(it) } + } - override fun delete(filters: List) { - val ids = HashSet() - query(filters) { ids.add(it.id) } - ids.forEach { delete(it) } - } + override fun delete(filters: List) = + lockManager.withWriteLock { + val ids = HashSet() + query(filters) { ids.add(it.id) } + ids.forEach { deleteLocked(it) } + } /** Delete an event by id. Returns 1 if a file was removed, 0 otherwise. */ - fun delete(id: HexKey): Int { + fun delete(id: HexKey): Int = + lockManager.withWriteLock { + deleteLocked(id) + } + + private fun deleteLocked(id: HexKey): Int { val canonical = layout.canonical(id) val event = readEvent(id) // need tags to know which index links to remove if (event != null) { @@ -366,21 +381,83 @@ class FsEventStore( * filenames, and deletes any entry whose `exp < now`. Matches SQLite's * `expiration < unixepoch()` predicate (note: strict `<`, not `<=`). */ - override fun deleteExpiredEvents() { - if (!Files.isDirectory(layout.idxExpiresAt)) return - val now = clock() - val toDelete = ArrayList() - Files.list(layout.idxExpiresAt).use { stream -> - for (entry in stream) { - val parsed = FsLayout.parseEntry(entry.fileName.toString()) ?: continue - if (parsed.first < now) toDelete.add(parsed.second) + override fun deleteExpiredEvents() = + lockManager.withWriteLock { + if (!Files.isDirectory(layout.idxExpiresAt)) return@withWriteLock + val now = clock() + val toDelete = ArrayList() + Files.list(layout.idxExpiresAt).use { stream -> + for (entry in stream) { + val parsed = FsLayout.parseEntry(entry.fileName.toString()) ?: continue + if (parsed.first < now) toDelete.add(parsed.second) + } } + toDelete.forEach { deleteLocked(it) } } - toDelete.forEach { delete(it) } - } override fun close() { - // No long-lived resources yet. + lockManager.close() + } + + // ------------------------------------------------------------------ + // Maintenance + // ------------------------------------------------------------------ + + /** + * Rebuild every `idx/` entry from the canonical events. Safe to run + * after external edits or to recover from a partial-write crash. + * Tombstones, replaceable / addressable slots, and the seed file are + * left untouched — slot-only-pinned events stay reachable, and + * tombstone removal is treated as a deliberate "un-forget" by the + * user. + */ + fun scrub() = + lockManager.withWriteLock { + cleanStaging() + // Wipe and recreate idx/. + deleteRecursively(layout.idx) + layout.ensureSkeleton() + + if (!Files.isDirectory(layout.events)) return@withWriteLock + Files.walk(layout.events).use { stream -> + for (path in stream) { + if (!Files.isRegularFile(path)) continue + if (!path.fileName.toString().endsWith(FsLayout.JSON_EXT)) continue + val event = + try { + Event.fromJson(Files.readString(path)) + } catch (_: Exception) { + continue + } + indexer.link(event, path) + } + } + } + + /** + * Drop dangling `idx/` entries whose canonical no longer exists. + * Cheaper than [scrub] because it never opens an event JSON; it just + * stats the canonical path encoded in the index entry filename. + */ + fun compact() = + lockManager.withWriteLock { + if (!Files.isDirectory(layout.idx)) return@withWriteLock + Files.walk(layout.idx).use { stream -> + for (path in stream) { + if (!Files.isRegularFile(path)) continue + val parsed = FsLayout.parseEntry(path.fileName.toString()) ?: continue + if (!layout.canonical(parsed.second).exists()) { + Files.deleteIfExists(path) + } + } + } + } + + private fun deleteRecursively(p: java.nio.file.Path) { + if (!Files.exists(p)) return + Files.walk(p).use { stream -> + stream.sorted(Comparator.reverseOrder()).forEach { Files.deleteIfExists(it) } + } } // ------------------------------------------------------------------ diff --git a/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsLockManager.kt b/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsLockManager.kt new file mode 100644 index 000000000..9371592c0 --- /dev/null +++ b/quartz/src/jvmMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsLockManager.kt @@ -0,0 +1,122 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip01Core.store.fs + +import java.nio.channels.FileChannel +import java.nio.channels.FileLock +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.StandardOpenOption + +/** + * Cross-process write serialisation for the file-backed event store. + * + * Acquires an exclusive `flock(2)` on `.lock` while the body runs, so a + * second `amy` invocation against the same directory blocks until the + * first releases. Re-entrant for the calling thread — nested + * `withWriteLock` calls (e.g. transaction body that calls insert) reuse + * the existing lock instead of self-deadlocking. + * + * Readers do not take this lock. The store's atomic-rename writes mean + * readers see either the pre- or post-mutation state, never a torn + * file. Stale `idx/` entries that point at a just-unlinked canonical + * are tolerated by the query loop's `NoSuchFileException` handling. + */ +internal class FsLockManager( + root: Path, +) : AutoCloseable { + private val lockPath: Path = root.resolve(LOCK_FILE) + private val mu = Any() + + /** Per-thread re-entry depth. Allows nested `withWriteLock` calls. */ + private val depth = ThreadLocal.withInitial { 0 } + + /** Active channel + lock when depth > 0 (any thread). Held by whoever owns the lock. */ + private var channel: FileChannel? = null + private var lock: FileLock? = null + + init { + try { + Files.createFile(lockPath) + } catch (_: java.nio.file.FileAlreadyExistsException) { + // existing lock file from a previous run — fine + } + } + + fun withWriteLock(body: () -> T): T { + // Re-entrant on the same thread: just run. + if (depth.get() > 0) { + depth.set(depth.get() + 1) + try { + return body() + } finally { + depth.set(depth.get() - 1) + } + } + + // Cross-thread + cross-process serialisation. + synchronized(mu) { + val ch = FileChannel.open(lockPath, StandardOpenOption.READ, StandardOpenOption.WRITE) + val l = ch.lock() // exclusive, blocking + channel = ch + lock = l + depth.set(1) + try { + return body() + } finally { + depth.set(0) + try { + l.release() + } catch (_: Throwable) { + // ignore + } + try { + ch.close() + } catch (_: Throwable) { + // ignore + } + channel = null + lock = null + } + } + } + + override fun close() { + synchronized(mu) { + try { + lock?.release() + } catch (_: Throwable) { + // ignore + } + try { + channel?.close() + } catch (_: Throwable) { + // ignore + } + channel = null + lock = null + } + } + + companion object { + const val LOCK_FILE = ".lock" + } +} diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsMaintenanceTest.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsMaintenanceTest.kt new file mode 100644 index 000000000..b5b004716 --- /dev/null +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/fs/FsMaintenanceTest.kt @@ -0,0 +1,258 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip01Core.store.fs + +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync +import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent +import com.vitorpamplona.quartz.utils.Secp256k1Instance +import java.nio.file.Files +import java.nio.file.Path +import kotlin.io.path.exists +import kotlin.io.path.listDirectoryEntries +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertTrue + +class FsMaintenanceTest { + private val signer = NostrSignerSync() + private lateinit var root: Path + private lateinit var store: FsEventStore + + @BeforeTest + fun setup() { + Secp256k1Instance + root = Files.createTempDirectory("fs-maint-") + store = FsEventStore(root) + } + + @AfterTest + fun tearDown() { + store.close() + if (root.exists()) { + Files.walk(root).use { it.sorted(Comparator.reverseOrder()).forEach { p -> Files.deleteIfExists(p) } } + } + } + + private fun note( + body: String, + ts: Long, + ) = signer.sign(TextNoteEvent.build(body, createdAt = ts)) + + // ------------------------------------------------------------------ + // Lock file + // ------------------------------------------------------------------ + + @Test + fun `lock file is created on open`() { + assertTrue(root.resolve(".lock").exists()) + } + + // ------------------------------------------------------------------ + // Transactions + // ------------------------------------------------------------------ + + @Test + fun `transaction commits all inserts on success`() { + val a = note("a", 1) + val b = note("b", 2) + val c = note("c", 3) + + store.transaction { + insert(a) + insert(b) + insert(c) + } + + val got = store.query(Filter(authors = listOf(signer.pubKey))) + assertEquals(setOf(a.id, b.id, c.id), got.map { it.id }.toSet()) + } + + @Test + fun `transaction propagates exceptions and stops processing`() { + val a = note("a", 1) + val b = note("b", 2) + val c = note("c", 3) + + assertFailsWith { + store.transaction { + insert(a) + insert(b) + throw IllegalStateException("boom") + // unreachable + @Suppress("UNREACHABLE_CODE") + insert(c) + } + } + + // Events written before the throw are kept (per the plan: atomic- + // per-event, serialised across writers — not all-or-nothing). + assertTrue(store.count(Filter(ids = listOf(a.id))) == 1) + assertTrue(store.count(Filter(ids = listOf(b.id))) == 1) + assertTrue(store.count(Filter(ids = listOf(c.id))) == 0) + } + + @Test + fun `transaction is re-entrant on the same thread`() { + val a = note("a", 1) + // If flock were non-reentrant we'd self-deadlock here because + // insert() acquires the same lock the transaction already holds. + store.transaction { + insert(a) + // Call an outer-locking method from within the transaction. + store.deleteExpiredEvents() + } + assertEquals(1, store.count(Filter(ids = listOf(a.id)))) + } + + // ------------------------------------------------------------------ + // scrub — rebuild idx/ from canonical + // ------------------------------------------------------------------ + + @Test + fun `scrub rebuilds idx entries after a manual wipe`() { + val a = note("hello bitcoin", 10) + val b = note("nostr stuff", 20) + store.insert(a) + store.insert(b) + + // Blow away the entire idx/ tree behind the store's back. + Files.walk(root.resolve("idx")).use { + it.sorted(Comparator.reverseOrder()).forEach { p -> Files.deleteIfExists(p) } + } + + // Without scrub, index-driven queries find nothing. + assertEquals(emptyList(), store.query(Filter(authors = listOf(signer.pubKey))).map { it.id }) + + store.scrub() + + val got = store.query(Filter(authors = listOf(signer.pubKey))).map { it.id }.toSet() + assertEquals(setOf(a.id, b.id), got) + + // FTS recovered too. + assertEquals(listOf(a.id), store.query(Filter(search = "bitcoin")).map { it.id }) + } + + @Test + fun `scrub leaves replaceable slot intact`() { + // Replaceable slots pin events via hardlink even without the + // canonical. Scrub must not wipe slots. + val meta = + signer.sign( + createdAt = 10, + kind = 0, + tags = emptyArray(), + content = "{}", + ) + store.insert(meta) + val slot = root.resolve("replaceable/0/${signer.pubKey}.json") + assertTrue(slot.exists()) + + store.scrub() + assertTrue(slot.exists(), "replaceable slot must survive scrub") + } + + // ------------------------------------------------------------------ + // compact — drop dangling idx entries + // ------------------------------------------------------------------ + + @Test + fun `compact drops idx entries whose canonical is gone`() { + val a = note("x", 10) + store.insert(a) + + // Externally delete the canonical without touching idx/. + val canonical = root.resolve("events/${a.id.substring(0, 2)}/${a.id.substring(2, 4)}/${a.id}.json") + assertTrue(Files.deleteIfExists(canonical)) + + val kindDir = root.resolve("idx/kind/1") + assertEquals(1, kindDir.listDirectoryEntries().size, "dangling entry still present pre-compact") + + store.compact() + + assertEquals(0, kindDir.listDirectoryEntries().size, "dangling entry dropped post-compact") + } + + @Test + fun `compact leaves valid entries alone`() { + val a = note("x", 10) + store.insert(a) + + store.compact() + + val kindDir = root.resolve("idx/kind/1") + assertEquals(1, kindDir.listDirectoryEntries().size, "valid entry should not be touched") + assertEquals(listOf(a.id), store.query(Filter(ids = listOf(a.id))).map { it.id }) + } + + // ------------------------------------------------------------------ + // close + // ------------------------------------------------------------------ + + @Test + fun `close is idempotent`() { + store.close() + store.close() + } + + @Test + fun `reopen after close works`() { + val a = note("a", 1) + store.insert(a) + store.close() + + val reopened = FsEventStore(root) + try { + assertEquals(listOf(a.id), reopened.query(Filter(ids = listOf(a.id))).map { it.id }) + } finally { + reopened.close() + } + } + + // ------------------------------------------------------------------ + // Concurrency — two writer threads serialise cleanly + // ------------------------------------------------------------------ + + @Test + fun `concurrent inserts on two threads are both persisted`() { + val events = (1..20).map { note("n$it", it.toLong()) } + val half = events.size / 2 + + val t1 = + Thread { + events.take(half).forEach { store.insert(it) } + } + val t2 = + Thread { + events.drop(half).forEach { store.insert(it) } + } + t1.start() + t2.start() + t1.join() + t2.join() + + val got = store.query(Filter(authors = listOf(signer.pubKey))).map { it.id }.toSet() + assertEquals(events.map { it.id }.toSet(), got) + } +}