chore(quartz): tighten fs store exception handling + lock manager
Audit cleanup — three correctness-adjacent fixes: #13 — FsLayout.readOrCreateSeed was TOCTOU-racy for two processes opening the same fresh directory. Both could pass `!exists()`, both write their own random bytes, both ATOMIC_MOVE into place. Loser's seed is forgotten but they already hashed with it, so their future index entries would be unreachable. Switch to CREATE_NEW: exactly one process creates the file, the other catches FileAlreadyExistsException and falls through to read the winner's bytes. #9 — `catch (_: Throwable)` / `catch (_: Exception)` in FsEventStore, FsSlots, FsTombstones narrowed to `catch (_: IOException)` plus `catch (_: JacksonException)` for JSON-parse paths. The old catches swallowed CancellationException (hides coroutine cancellation), OutOfMemoryError, and ThreadDeath. Now only the errors we actually expect on a racing read are suppressed. #1 — FsLockManager replaced synchronized+ThreadLocal+depth-counter with a ReentrantLock. Same semantics (cross-process flock + in-process reentry), cleaner code: the ReentrantLock handles reentry natively via holdCount, no ThreadLocal, no manual depth bookkeeping. Release is guarded by `holdCount == 1` on exit. Narrow catches in the cleanup path to IOException too. 113 fs tests green across all suites.
This commit is contained in:
+3
-1
@@ -429,7 +429,9 @@ open class FsEventStore(
|
|||||||
val event =
|
val event =
|
||||||
try {
|
try {
|
||||||
Event.fromJson(Files.readString(path))
|
Event.fromJson(Files.readString(path))
|
||||||
} catch (_: Exception) {
|
} catch (_: java.io.IOException) {
|
||||||
|
continue
|
||||||
|
} catch (_: com.fasterxml.jackson.core.JacksonException) {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
indexer.link(event, path)
|
indexer.link(event, path)
|
||||||
|
|||||||
@@ -27,7 +27,7 @@ import com.vitorpamplona.quartz.utils.sha256.sha256
|
|||||||
import java.nio.ByteBuffer
|
import java.nio.ByteBuffer
|
||||||
import java.nio.file.Files
|
import java.nio.file.Files
|
||||||
import java.nio.file.Path
|
import java.nio.file.Path
|
||||||
import java.nio.file.StandardCopyOption
|
import java.nio.file.StandardOpenOption
|
||||||
import java.security.SecureRandom
|
import java.security.SecureRandom
|
||||||
import kotlin.io.path.exists
|
import kotlin.io.path.exists
|
||||||
|
|
||||||
@@ -162,15 +162,29 @@ internal class FsLayout(
|
|||||||
* Read the seed salt (creates it on first call). The seed is
|
* Read the seed salt (creates it on first call). The seed is
|
||||||
* write-once — reopening an existing store must return the same Long
|
* write-once — reopening an existing store must return the same Long
|
||||||
* or every previously-written index entry becomes unreachable.
|
* or every previously-written index entry becomes unreachable.
|
||||||
|
*
|
||||||
|
* Concurrent first-open: two processes opening the same fresh
|
||||||
|
* directory simultaneously must not both write-then-use their own
|
||||||
|
* random seed. We use `CREATE_NEW` so exactly one process creates
|
||||||
|
* the file; any loser catches `FileAlreadyExistsException` and
|
||||||
|
* falls through to read the winner's bytes.
|
||||||
*/
|
*/
|
||||||
fun readOrCreateSeed(): Long {
|
fun readOrCreateSeed(): Long {
|
||||||
val f = root.resolve(SEED_FILE)
|
val f = root.resolve(SEED_FILE)
|
||||||
if (!f.exists()) {
|
if (!f.exists()) {
|
||||||
val bytes = ByteArray(8)
|
val bytes = ByteArray(8)
|
||||||
SecureRandom().nextBytes(bytes)
|
SecureRandom().nextBytes(bytes)
|
||||||
val tmp = Files.createTempFile(root, ".seed-", ".tmp")
|
try {
|
||||||
Files.write(tmp, bytes)
|
Files
|
||||||
Files.move(tmp, f, StandardCopyOption.ATOMIC_MOVE)
|
.newByteChannel(
|
||||||
|
f,
|
||||||
|
setOf(StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE),
|
||||||
|
).use { ch ->
|
||||||
|
ch.write(ByteBuffer.wrap(bytes))
|
||||||
|
}
|
||||||
|
} catch (_: java.nio.file.FileAlreadyExistsException) {
|
||||||
|
// Another process won the race; fall through to read theirs.
|
||||||
|
}
|
||||||
}
|
}
|
||||||
val bytes = Files.readAllBytes(f)
|
val bytes = Files.readAllBytes(f)
|
||||||
require(bytes.size == 8) { "$SEED_FILE must be exactly 8 bytes, got ${bytes.size}" }
|
require(bytes.size == 8) { "$SEED_FILE must be exactly 8 bytes, got ${bytes.size}" }
|
||||||
|
|||||||
+52
-49
@@ -20,38 +20,44 @@
|
|||||||
*/
|
*/
|
||||||
package com.vitorpamplona.quartz.nip01Core.store.fs
|
package com.vitorpamplona.quartz.nip01Core.store.fs
|
||||||
|
|
||||||
|
import java.io.IOException
|
||||||
import java.nio.channels.FileChannel
|
import java.nio.channels.FileChannel
|
||||||
import java.nio.channels.FileLock
|
import java.nio.channels.FileLock
|
||||||
import java.nio.file.Files
|
import java.nio.file.Files
|
||||||
import java.nio.file.Path
|
import java.nio.file.Path
|
||||||
import java.nio.file.StandardOpenOption
|
import java.nio.file.StandardOpenOption
|
||||||
|
import java.util.concurrent.locks.ReentrantLock
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Cross-process write serialisation for the file-backed event store.
|
* Cross-process write serialisation for the file-backed event store.
|
||||||
*
|
*
|
||||||
* Acquires an exclusive `flock(2)` on `.lock` while the body runs, so a
|
* Two layers of mutual exclusion:
|
||||||
* 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
|
* - [inProcessLock] — a `ReentrantLock` serialising threads within this
|
||||||
* readers see either the pre- or post-mutation state, never a torn
|
* JVM. Reentrant on the same thread, so nested
|
||||||
* file. Stale `idx/` entries that point at a just-unlinked canonical
|
* `withWriteLock` calls (e.g. a transaction body that calls insert)
|
||||||
* are tolerated by the query loop's `NoSuchFileException` handling.
|
* reuse the lock instead of self-deadlocking.
|
||||||
|
*
|
||||||
|
* - [FileChannel.lock] on `.lock` — a POSIX/Windows advisory file lock
|
||||||
|
* serialising this JVM against other JVMs pointed at the same store.
|
||||||
|
* Acquired once on first entry, released once on last exit (tracked
|
||||||
|
* via the reentrant hold count).
|
||||||
|
*
|
||||||
|
* Readers do not take either 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(
|
internal class FsLockManager(
|
||||||
root: Path,
|
root: Path,
|
||||||
) : AutoCloseable {
|
) : AutoCloseable {
|
||||||
private val lockPath: Path = root.resolve(LOCK_FILE)
|
private val lockPath: Path = root.resolve(LOCK_FILE)
|
||||||
private val mu = Any()
|
private val inProcessLock = ReentrantLock()
|
||||||
|
|
||||||
/** Per-thread re-entry depth. Allows nested `withWriteLock` calls. */
|
/** Held only while this JVM owns the file lock (depth ≥ 1). Guarded by [inProcessLock]. */
|
||||||
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 channel: FileChannel? = null
|
||||||
private var lock: FileLock? = null
|
private var fileLock: FileLock? = null
|
||||||
|
|
||||||
init {
|
init {
|
||||||
try {
|
try {
|
||||||
@@ -62,58 +68,55 @@ internal class FsLockManager(
|
|||||||
}
|
}
|
||||||
|
|
||||||
fun <T> withWriteLock(body: () -> T): T {
|
fun <T> withWriteLock(body: () -> T): T {
|
||||||
// Re-entrant on the same thread: just run.
|
inProcessLock.lock()
|
||||||
if (depth.get() > 0) {
|
|
||||||
depth.set(depth.get() + 1)
|
|
||||||
try {
|
try {
|
||||||
return body()
|
// Only the outermost re-entry actually touches the file lock.
|
||||||
} finally {
|
if (inProcessLock.holdCount == 1) {
|
||||||
depth.set(depth.get() - 1)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Cross-thread + cross-process serialisation.
|
|
||||||
synchronized(mu) {
|
|
||||||
val ch = FileChannel.open(lockPath, StandardOpenOption.READ, StandardOpenOption.WRITE)
|
val ch = FileChannel.open(lockPath, StandardOpenOption.READ, StandardOpenOption.WRITE)
|
||||||
val l = ch.lock() // exclusive, blocking
|
val l =
|
||||||
|
try {
|
||||||
|
ch.lock() // exclusive, blocking
|
||||||
|
} catch (t: Throwable) {
|
||||||
|
ch.close()
|
||||||
|
throw t
|
||||||
|
}
|
||||||
channel = ch
|
channel = ch
|
||||||
lock = l
|
fileLock = l
|
||||||
depth.set(1)
|
}
|
||||||
try {
|
try {
|
||||||
return body()
|
return body()
|
||||||
} finally {
|
} finally {
|
||||||
depth.set(0)
|
if (inProcessLock.holdCount == 1) {
|
||||||
try {
|
releaseFileLock()
|
||||||
l.release()
|
|
||||||
} catch (_: Throwable) {
|
|
||||||
// ignore
|
|
||||||
}
|
}
|
||||||
try {
|
|
||||||
ch.close()
|
|
||||||
} catch (_: Throwable) {
|
|
||||||
// ignore
|
|
||||||
}
|
|
||||||
channel = null
|
|
||||||
lock = null
|
|
||||||
}
|
}
|
||||||
|
} finally {
|
||||||
|
inProcessLock.unlock()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
override fun close() {
|
override fun close() {
|
||||||
synchronized(mu) {
|
inProcessLock.lock()
|
||||||
try {
|
try {
|
||||||
lock?.release()
|
releaseFileLock()
|
||||||
} catch (_: Throwable) {
|
} finally {
|
||||||
// ignore
|
inProcessLock.unlock()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun releaseFileLock() {
|
||||||
|
try {
|
||||||
|
fileLock?.release()
|
||||||
|
} catch (_: IOException) {
|
||||||
|
// best-effort during unwind
|
||||||
}
|
}
|
||||||
try {
|
try {
|
||||||
channel?.close()
|
channel?.close()
|
||||||
} catch (_: Throwable) {
|
} catch (_: IOException) {
|
||||||
// ignore
|
// best-effort during unwind
|
||||||
}
|
}
|
||||||
|
fileLock = null
|
||||||
channel = null
|
channel = null
|
||||||
lock = null
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
companion object {
|
companion object {
|
||||||
|
|||||||
@@ -93,9 +93,9 @@ internal class FsSlots(
|
|||||||
if (!slot.exists()) return null
|
if (!slot.exists()) return null
|
||||||
return try {
|
return try {
|
||||||
Event.fromJson(slot.readText())
|
Event.fromJson(slot.readText())
|
||||||
} catch (_: java.nio.file.NoSuchFileException) {
|
} catch (_: java.io.IOException) {
|
||||||
null
|
null
|
||||||
} catch (_: Exception) {
|
} catch (_: com.fasterxml.jackson.core.JacksonException) {
|
||||||
null
|
null
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+4
-4
@@ -70,9 +70,9 @@ internal class FsTombstones(
|
|||||||
if (!path.exists()) return null
|
if (!path.exists()) return null
|
||||||
return try {
|
return try {
|
||||||
Event.fromJson(path.readText()).createdAt
|
Event.fromJson(path.readText()).createdAt
|
||||||
} catch (_: java.nio.file.NoSuchFileException) {
|
} catch (_: java.io.IOException) {
|
||||||
null
|
null
|
||||||
} catch (_: Exception) {
|
} catch (_: com.fasterxml.jackson.core.JacksonException) {
|
||||||
null
|
null
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -135,9 +135,9 @@ internal class FsTombstones(
|
|||||||
if (!path.exists()) return null
|
if (!path.exists()) return null
|
||||||
return try {
|
return try {
|
||||||
Event.fromJson(path.readText()).createdAt
|
Event.fromJson(path.readText()).createdAt
|
||||||
} catch (_: java.nio.file.NoSuchFileException) {
|
} catch (_: java.io.IOException) {
|
||||||
null
|
null
|
||||||
} catch (_: Exception) {
|
} catch (_: com.fasterxml.jackson.core.JacksonException) {
|
||||||
null
|
null
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user