From bc7e35b54e9532f6345b110c07b3ac29a8eb05a9 Mon Sep 17 00:00:00 2001 From: Renn F Date: Wed, 14 Oct 2026 22:54:06 +0200 Subject: [PATCH] fix(agent): persist to Redis off the buffer lock and drain pending writes on stop --- .../kotlin/com/guardcore/agent/Encryption.kt | 3 +- .../kotlin/com/guardcore/agent/GuardAgent.kt | 165 ++++++++++--- .../com/guardcore/agent/CoverageMoreTest.kt | 12 +- .../com/guardcore/agent/EncryptionTest.kt | 6 +- .../com/guardcore/agent/PersistOffLockTest.kt | 219 ++++++++++++++++++ .../agent/PersistenceIntegrationTest.kt | 4 + .../kotlin/com/guardcore/agent/TestSupport.kt | 8 + 7 files changed, 368 insertions(+), 49 deletions(-) create mode 100644 src/test/kotlin/com/guardcore/agent/PersistOffLockTest.kt diff --git a/src/main/kotlin/com/guardcore/agent/Encryption.kt b/src/main/kotlin/com/guardcore/agent/Encryption.kt index 7c84a52..740b1c7 100644 --- a/src/main/kotlin/com/guardcore/agent/Encryption.kt +++ b/src/main/kotlin/com/guardcore/agent/Encryption.kt @@ -29,7 +29,8 @@ public class EncryptionError( */ public class EncryptionConfigError( message: String, -) : RuntimeException(message) + cause: Throwable? = null, +) : RuntimeException(message, cause) /** encryption.go `PayloadEncryptor`: encrypts telemetry payloads with AES-256-GCM. */ public class PayloadEncryptor diff --git a/src/main/kotlin/com/guardcore/agent/GuardAgent.kt b/src/main/kotlin/com/guardcore/agent/GuardAgent.kt index 9d56672..07d114b 100644 --- a/src/main/kotlin/com/guardcore/agent/GuardAgent.kt +++ b/src/main/kotlin/com/guardcore/agent/GuardAgent.kt @@ -73,6 +73,12 @@ internal constructor( /** confirmQueue serializes Redis confirmations on one worker thread. */ internal val confirmQueue = LinkedBlockingQueue>(64) + /** + * Redis persists run here, never on the caller's thread and never under [mutex]; [stop] waits for + * them. Null when persistence is not configured, so nothing is started then. + */ + internal var persistPool: java.util.concurrent.ThreadPoolExecutor? = null + @Volatile private var confirmClosed: Boolean = false private val workerDone = CountDownLatch(1) private val loopThreads = mutableListOf() @@ -84,12 +90,14 @@ internal constructor( /** agent.go `bufferedEvent`: one queued event plus its persistence key. */ internal class BufferedEvent( val ev: SecurityEvent, - val key: String, + @Volatile var key: String, + val reserved: String = "", ) internal class BufferedMetric( val m: SecurityMetric, - val key: String, + @Volatile var key: String, + val reserved: String = "", ) /** agent.go `Status`: a point-in-time snapshot of the agent's health and counters. */ @@ -154,6 +162,9 @@ internal constructor( internal const val DEGRADED_FAILURE_RATE: Double = 0.1 internal const val HEALTH_FAILURE_RATE_MAX: Double = 0.5 internal const val MAX_TRACKED_ERRORS: Int = 10 + internal const val PERSIST_THREADS: Int = 4 + internal const val PERSIST_QUEUE: Int = 1024 + internal const val PERSIST_DRAIN_MS: Long = 10_000 internal const val KIND_EVENT: String = "event" internal const val KIND_METRIC: String = "metric" internal const val ENGINE_CLASS_NAME: String = "com.guardcore.engine.GuardCoreEngine" @@ -188,7 +199,9 @@ internal constructor( createEncryptor(cfg.projectEncryptionKey) } catch (encErr: Exception) { throw EncryptionConfigError( - "Encryption round-trip failed at startup; refusing plaintext fallback", + "Encryption round-trip failed at startup; refusing plaintext fallback " + + "(cause: ${encErr.message ?: encErr::class.java.simpleName})", + encErr, ) } if (encryptor != null && !encryptor.verifyKey()) { @@ -212,6 +225,16 @@ internal constructor( ) } agent.persist = persist + agent.persistPool = + java.util.concurrent.ThreadPoolExecutor( + PERSIST_THREADS, + PERSIST_THREADS, + 0L, + TimeUnit.MILLISECONDS, + LinkedBlockingQueue(PERSIST_QUEUE), + ) { r -> + Thread(r, "guard-agent-persist").apply { isDaemon = true } + } val worker = Thread({ agent.confirmWorker() }, "guard-agent-confirm") worker.isDaemon = true worker.start() @@ -268,6 +291,10 @@ internal constructor( for (t in threads) { t.join(10_000) } + persistPool?.let { + it.shutdown() + it.awaitTermination(PERSIST_DRAIN_MS, TimeUnit.MILLISECONDS) + } val errE = flushEvents(ctx, force = true) val errM = flushMetrics(ctx, force = true) val err = joinErrors(errE, errM) @@ -718,9 +745,11 @@ internal constructor( var evicted: BufferedEvent? = null var droppedCount = 0L var needFlush = false + var item: BufferedEvent? = null synchronized(mutex) { if (events.size < cfg.bufferSize) { - needFlush = appendEventLocked(ev) + item = appendEventLocked(ev) + needFlush = watermarkReachedLocked() action = 1 } else { when (cfg.overflow) { @@ -729,7 +758,8 @@ internal constructor( val old = events.removeFirst() eventsDropped++ droppedCount = eventsDropped - needFlush = appendEventLocked(ev) + item = appendEventLocked(ev) + needFlush = watermarkReachedLocked() evicted = old action = 2 } @@ -740,12 +770,14 @@ internal constructor( } when (action) { 1 -> { + spawnPersistEvent(item!!) if (needFlush) { triggerFlush() } return } 2 -> { + spawnPersistEvent(item!!) logDrop(KIND_EVENT, droppedCount) evicted?.key?.takeIf { it.isNotEmpty() }?.let { confirmKeys(listOf(it)) } if (needFlush) { @@ -767,9 +799,11 @@ internal constructor( var evicted: BufferedMetric? = null var droppedCount = 0L var needFlush = false + var item: BufferedMetric? = null synchronized(mutex) { if (metrics.size < cfg.bufferSize) { - needFlush = appendMetricLocked(m) + item = appendMetricLocked(m) + needFlush = watermarkReachedLocked() action = 1 } else { when (cfg.overflow) { @@ -778,7 +812,8 @@ internal constructor( val old = metrics.removeFirst() metricsDropped++ droppedCount = metricsDropped - needFlush = appendMetricLocked(m) + item = appendMetricLocked(m) + needFlush = watermarkReachedLocked() evicted = old action = 2 } @@ -789,12 +824,14 @@ internal constructor( } when (action) { 1 -> { + spawnPersistMetric(item!!) if (needFlush) { triggerFlush() } return } 2 -> { + spawnPersistMetric(item!!) logDrop(KIND_METRIC, droppedCount) evicted?.key?.takeIf { it.isNotEmpty() }?.let { confirmKeys(listOf(it)) } if (needFlush) { @@ -827,63 +864,115 @@ internal constructor( } /** - * agent.go `appendEventLocked`: appends under a held lock, persisting to Redis first - * (persist-on-enqueue, mirroring the Python agent which awaits the write while holding the buffer - * lock). It returns whether the combined occupancy reached the high-watermark. + * agent.go `appendEventLocked`: appends under a held lock. Redis is never touched here: the + * caller hands the returned item to [spawnPersistEvent] after releasing the lock. */ - private fun appendEventLocked(ev: SecurityEvent): Boolean { - val key = reserveEventKeyLocked(ev) - events.add(BufferedEvent(ev, key)) + private fun appendEventLocked(ev: SecurityEvent): BufferedEvent { + // The key is minted here, under the lock, so key order equals append order (reload sorts keys). + val item = BufferedEvent(ev, "", mintKey(PERSIST_NAMESPACE_EVENTS, "event")) + events.add(item) eventsBuffered++ - return watermarkReachedLocked() + return item } - private fun appendMetricLocked(m: SecurityMetric): Boolean { - val key = reserveMetricKeyLocked(m) - metrics.add(BufferedMetric(m, key)) + private fun appendMetricLocked(m: SecurityMetric): BufferedMetric { + val item = BufferedMetric(m, "", mintKey(PERSIST_NAMESPACE_METRICS, "metric")) + metrics.add(item) metricsBuffered++ - return watermarkReachedLocked() + return item } + private fun mintKey( + namespace: String, + kind: String, + ): String = persist?.let { it.fullKey(namespace, it.shortKey(kind)) } ?: "" + private fun watermarkReachedLocked(): Boolean { val combined = events.size + metrics.size return combined.toDouble() >= cfg.bufferSize.toDouble() * cfg.highWatermarkRatio } - internal fun reserveEventKeyLocked(ev: SecurityEvent): String { - val p = persist ?: return "" + /** Runs [task] on the persist pool; a full or closed pool skips persistence (fail-open). */ + private fun submitPersist(task: () -> Unit) { + val pool = persistPool ?: return + try { + pool.execute { + try { + task() + } catch (e: Throwable) { + logger.warning("guardagent: redis persist task failed: $e") + } + } + } catch (e: java.util.concurrent.RejectedExecutionException) { + synchronized(mutex) { redisPersistFailures++ } + } + } + + private fun spawnPersistEvent(item: BufferedEvent) = submitPersist { persistEventNow(item) } + + private fun spawnPersistMetric(item: BufferedMetric) = submitPersist { persistMetricNow(item) } + + /** Persists one buffered event and settles its key. Never called with [mutex] held. */ + internal fun persistEventNow(item: BufferedEvent) { val data = try { - goJson(ev) + goJson(item.ev) } catch (err: Exception) { - redisPersistFailures++ - recordErrorLocked("event serialization for persistence failed: ${err.message}") - return "" + synchronized(mutex) { + redisPersistFailures++ + recordErrorLocked("event serialization for persistence failed: ${err.message}") + } + return } - return persistItemLocked(PERSIST_NAMESPACE_EVENTS, "event", data) + val key = + persistItem(item.reserved.ifEmpty { mintKey(PERSIST_NAMESPACE_EVENTS, "event") }, data) + if (key.isEmpty()) { + return + } + // Still buffered (including after a requeue): track the key. Flushed or evicted before + // the key came back: delete it so it neither orphans nor replays after a restart. + val tracked = + synchronized(mutex) { + val buffered = events.asReversed().any { it === item } + if (buffered) item.key = key + buffered + } + if (!tracked) { + confirmKeys(listOf(key)) + } } - internal fun reserveMetricKeyLocked(m: SecurityMetric): String { - val p = persist ?: return "" - // SecurityMetric is JSON-serializable by construction (scalar and - // string-map fields only), so unlike events its serialization cannot - // fail; only the persistence write can. - val data = goJson(m) - return persistItemLocked(PERSIST_NAMESPACE_METRICS, "metric", data) + /** Metric twin of [persistEventNow]. */ + internal fun persistMetricNow(item: BufferedMetric) { + val key = + persistItem( + item.reserved.ifEmpty { mintKey(PERSIST_NAMESPACE_METRICS, "metric") }, + goJson(item.m), + ) + if (key.isEmpty()) { + return + } + val tracked = + synchronized(mutex) { + val buffered = metrics.asReversed().any { it === item } + if (buffered) item.key = key + buffered + } + if (!tracked) { + confirmKeys(listOf(key)) + } } - private fun persistItemLocked( - namespace: String, - kind: String, + /** Writes one record to Redis. Must be called WITHOUT [mutex] held. */ + private fun persistItem( + key: String, data: ByteArray, ): String { val p = persist!! - val short = p.shortKey(kind) - val key = p.fullKey(namespace, short) if (p.store(key, data.toString(Charsets.UTF_8))) { return key } - redisPersistFailures++ + synchronized(mutex) { redisPersistFailures++ } return "" } diff --git a/src/test/kotlin/com/guardcore/agent/CoverageMoreTest.kt b/src/test/kotlin/com/guardcore/agent/CoverageMoreTest.kt index c047883..d4adca3 100644 --- a/src/test/kotlin/com/guardcore/agent/CoverageMoreTest.kt +++ b/src/test/kotlin/com/guardcore/agent/CoverageMoreTest.kt @@ -346,15 +346,11 @@ class CoverageMoreTest { agent.persist = deadPersist val poisoned = SecurityEvent(null, "e", metadata = linkedMapOf("chan" to Thread())) - var key: String - synchronized(agent.mutex) { key = agent.reserveEventKeyLocked(poisoned) } - assertEquals("", key, "an unserializable event must not reserve a key") + agent.persistEventNow(GuardAgent.BufferedEvent(poisoned, "")) assertEquals(1L, agent.stats().redisPersistFailures, "serialization failures must count") - var metricKey: String - synchronized(agent.mutex) { - metricKey = agent.reserveMetricKeyLocked(SecurityMetric(null, MetricTypes.REQUEST_COUNT)) - } - assertEquals("", metricKey, "a failing store must not reserve a metric key") + agent.persistMetricNow( + GuardAgent.BufferedMetric(SecurityMetric(null, MetricTypes.REQUEST_COUNT), ""), + ) assertEquals(2L, agent.stats().redisPersistFailures, "store failures must count") deadPersist.close() } diff --git a/src/test/kotlin/com/guardcore/agent/EncryptionTest.kt b/src/test/kotlin/com/guardcore/agent/EncryptionTest.kt index f96fb03..d2d1893 100644 --- a/src/test/kotlin/com/guardcore/agent/EncryptionTest.kt +++ b/src/test/kotlin/com/guardcore/agent/EncryptionTest.kt @@ -147,8 +147,10 @@ class EncryptionTest { cfg.apiKey = "test-api-key-123" cfg.projectEncryptionKey = "not-a-valid-key" val err = assertFailsWith { GuardAgent.create(cfg) } - assertEquals( - "Encryption round-trip failed at startup; refusing plaintext fallback", + assertTrue( + err.message!!.startsWith( + "Encryption round-trip failed at startup; refusing plaintext fallback", + ), err.message, ) } diff --git a/src/test/kotlin/com/guardcore/agent/PersistOffLockTest.kt b/src/test/kotlin/com/guardcore/agent/PersistOffLockTest.kt new file mode 100644 index 0000000..e64c10b --- /dev/null +++ b/src/test/kotlin/com/guardcore/agent/PersistOffLockTest.kt @@ -0,0 +1,219 @@ +package com.guardcore.agent + +import java.time.Duration +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import kotlin.test.AfterTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * Redis persistence runs off the request path and off the buffer mutex. Slow writes are gated on + * latches (the Persistence command fault injector blocks inside the SET), never on sleeps. Needs + * Redis on REDIS_HOST:6379, like PersistenceIntegrationTest; skips when unreachable. + */ +class PersistOffLockTest { + private val closers = mutableListOf() + private val wait = Duration.ofSeconds(15) + + @AfterTest + fun cleanup() { + for (c in closers.reversed()) { + try { + c.close() + } catch (ignored: Exception) {} + } + closers.clear() + } + + private fun host(): String? { + val h = System.getenv("REDIS_HOST") ?: "127.0.0.1" + return try { + val client = io.lettuce.core.RedisClient.create("redis://$h:6379") + val conn = client.connect() + val pong = conn.sync().ping() + conn.close() + client.shutdown() + if (pong == "PONG") h else null + } catch (e: Exception) { + null + } + } + + private class Rig( + val agent: GuardAgent, + val mock: MockIngest, + val prefix: String, + val host: String, + val entered: CountDownLatch, + val gate: CountDownLatch, + ) { + fun keys(): List { + val client = io.lettuce.core.RedisClient.create("redis://$host:6379") + val conn = client.connect() + try { + return conn.sync().keys("$prefix:*") + } finally { + conn.close() + client.shutdown() + } + } + } + + /** An agent whose first [blocked] SET calls park on `gate` until the test opens it. */ + private fun rig(blocked: Int = 1): Rig? { + val h = host() ?: return null + val mock = MockIngest() + closers.add(mock) + val prefix = "guardagent-offlock-" + System.nanoTime() + val entered = CountDownLatch(blocked) + val gate = CountDownLatch(1) + val seen = AtomicInteger() + val agent = + newTestAgent(tempDirFor("offlock"), mock) { cfg -> + cfg.retryAttempts = 0 + cfg.redis = RedisConfig("redis://$h:6379", prefix, Duration.ofMinutes(1)) + } + agent.persist!!.commandFaultInjector = { name -> + if (name == "set" && seen.incrementAndGet() <= blocked) { + entered.countDown() + gate.await(30, TimeUnit.SECONDS) + } + null + } + val rig = Rig(agent, mock, prefix, h, entered, gate) + closers.add( + AutoCloseable { + gate.countDown() + agent.stop() + val client = io.lettuce.core.RedisClient.create("redis://$h:6379") + val conn = client.connect() + val ks = conn.sync().keys("$prefix:*") + if (ks.isNotEmpty()) conn.sync().del(*ks.toTypedArray()) + conn.close() + client.shutdown() + }, + ) + return rig + } + + private fun inThread(block: () -> Unit): CountDownLatch { + val done = CountDownLatch(1) + Thread { + try { + block() + } finally { + done.countDown() + } + } + .apply { isDaemon = true } + .start() + return done + } + + private fun awaitPersistDone(r: Rig, tasks: Long) { + waitUntil(wait, "persist tasks complete") { + r.agent.persistPool!!.completedTaskCount >= tasks + } + } + + @Test + fun sendReturnsBeforeSlowWriteCompletes() { + val r = rig() ?: return + val sent = inThread { r.agent.sendEvent(testEvent(1)) } + assertTrue(r.entered.await(15, TimeUnit.SECONDS), "write never started") + assertTrue(sent.await(15, TimeUnit.SECONDS), "sendEvent must not wait for the Redis write") + r.gate.countDown() + awaitPersistDone(r, 1) + waitUntil(wait, "key tracked") { + synchronized(r.agent.mutex) { r.agent.events[0].key.isNotEmpty() } + } + assertEquals(1, r.keys().size) + } + + @Test + fun concurrentSendsAreNotSerializedBehindSlowWrite() { + val r = rig() ?: return + val first = inThread { r.agent.sendEvent(testEvent(1)) } + assertTrue(r.entered.await(15, TimeUnit.SECONDS)) + val second = inThread { r.agent.sendEvent(testEvent(2)) } + val third = inThread { r.agent.sendMetric(testMetric(3)) } + assertTrue(first.await(15, TimeUnit.SECONDS)) + assertTrue(second.await(15, TimeUnit.SECONDS), "second send blocked behind the slow write") + assertTrue(third.await(15, TimeUnit.SECONDS), "metric send blocked behind the slow write") + assertTrue( + synchronized(r.agent.mutex) { r.agent.events.size == 2 && r.agent.metrics.size == 1 }, + ) + r.gate.countDown() + awaitPersistDone(r, 3) + } + + @Test + fun stopDrainsPendingPersists() { + val r = rig() ?: return + r.agent.sendEvent(testEvent(1)) + assertTrue(r.entered.await(15, TimeUnit.SECONDS)) + val stopped = inThread { r.agent.stop() } + assertFalse(stopped.await(200, TimeUnit.MILLISECONDS), "stop returned with a persist in flight") + r.gate.countDown() + assertTrue(stopped.await(30, TimeUnit.SECONDS), "stop never finished") + // The write landed before the final flush, which then confirmed (deleted) it. + waitUntil(wait, "key confirmed after final flush") { r.keys().isEmpty() } + assertEquals(1, r.mock.allCalls().count { it.events.isNotEmpty() }) + } + + @Test + fun flushDuringPersistDeletesTheKey() { + val r = rig() ?: return + r.agent.sendEvent(testEvent(1)) + assertTrue(r.entered.await(15, TimeUnit.SECONDS)) + assertNull(r.agent.flush()) + r.gate.countDown() + awaitPersistDone(r, 1) + waitUntil(wait, "late key deleted, not orphaned") { r.keys().isEmpty() } + } + + @Test + fun requeueDuringPersistTracksTheKey() { + val r = rig() ?: return + r.agent.sendEvent(testEvent(1)) + assertTrue(r.entered.await(15, TimeUnit.SECONDS)) + r.mock.failNext(1) + assertNotNull(r.agent.flush(), "flush should fail and requeue") + assertEquals(1, synchronized(r.agent.mutex) { r.agent.events.size }) + r.gate.countDown() + awaitPersistDone(r, 1) + waitUntil(wait, "key tracked on the requeued item") { + synchronized(r.agent.mutex) { r.agent.events[0].key.isNotEmpty() } + } + assertEquals(1, r.keys().size) + } + + @Test + fun nothingStartedWithoutPersistence() { + val agent = newTestAgent(tempDirFor("nopersist"), null) + closers.add(AutoCloseable { agent.stop() }) + assertNull(agent.persistPool) + agent.sendEvent(testEvent(1)) + agent.sendMetric(testMetric(1)) + assertEquals(1, synchronized(agent.mutex) { agent.events.size }) + assertEquals("", synchronized(agent.mutex) { agent.events[0].key }) + } + + @Test + fun startupEncryptionErrorIncludesCause() { + val cfg = defaultConfig() + cfg.apiKey = "test-api-key-123" + cfg.projectEncryptionKey = "not-a-valid-key" + val err = kotlin.test.assertFailsWith { GuardAgent.create(cfg) } + assertTrue(err.message!!.contains("refusing plaintext fallback"), err.message) + assertTrue(err.message!!.contains("(cause: "), err.message) + assertNotNull(err.cause) + assertTrue(err.message!!.contains(err.cause!!.message ?: err.cause!!::class.java.simpleName)) + } +} diff --git a/src/test/kotlin/com/guardcore/agent/PersistenceIntegrationTest.kt b/src/test/kotlin/com/guardcore/agent/PersistenceIntegrationTest.kt index 04f14f7..93f6ad2 100644 --- a/src/test/kotlin/com/guardcore/agent/PersistenceIntegrationTest.kt +++ b/src/test/kotlin/com/guardcore/agent/PersistenceIntegrationTest.kt @@ -461,6 +461,7 @@ class PersistenceIntegrationTest { val agent = newIntegrationAgent(host, mock.url) try { for (i in 1..2) agent.sendEvent(testEvent(i)) + agent.awaitPersists() val keys = scanAll(client, INTEGRATION_REDIS_PREFIX + ":*") assertEquals(2, keys.size, "persisted keys") for (key in keys) { @@ -494,6 +495,7 @@ class PersistenceIntegrationTest { val agent = newIntegrationAgent(host, mock.url) try { agent.sendMetric(testMetric(1)) + agent.awaitPersists() assertEquals( 1, scanAll(client, INTEGRATION_REDIS_PREFIX + ":*").size, @@ -524,6 +526,7 @@ class PersistenceIntegrationTest { val first = newIntegrationAgent(host, "http://127.0.0.1:1") run { for (i in 1..2) first.sendEvent(testEvent(i)) + first.awaitPersists() assertTrue(first.flush() != null, "flush against a down endpoint must fail") assertTrue( first.stop() != null, @@ -606,6 +609,7 @@ class PersistenceIntegrationTest { mock.callsTo(EVENTS_PATH).size, "telemetry must still be delivered without redis", ) + agent.awaitPersists() val stats = agent.stats() assertTrue(stats.redisPersistFailures > 0, "redis failures must be counted: $stats") assertTrue(stats.durabilityDegraded, "durability must be reported degraded") diff --git a/src/test/kotlin/com/guardcore/agent/TestSupport.kt b/src/test/kotlin/com/guardcore/agent/TestSupport.kt index a5d6b43..47c730c 100644 --- a/src/test/kotlin/com/guardcore/agent/TestSupport.kt +++ b/src/test/kotlin/com/guardcore/agent/TestSupport.kt @@ -108,3 +108,11 @@ class AgentTestPanicDoer : HttpDoer { ctx: AgentContext, ): HttpResponse = throw IllegalStateException("transport panic") } + +/** Blocks until every queued Redis persist has completed (persistence is asynchronous). */ +fun GuardAgent.awaitPersists() { + val pool = persistPool ?: return + waitUntil(Duration.ofSeconds(15), "pending persists drained") { + pool.taskCount == pool.completedTaskCount + } +}