Serve the JDBC metastore's rows again, for as long as it exists

engine/src/main/kotlin/com/kakao/actionbase/v2/engine/metastore/JdbcMetastoreInspector.kt

@@ -0,0 +1,88 @@
+package com.kakao.actionbase.v2.engine.metastore
+
+import com.kakao.actionbase.v2.core.code.DecodedEdge
+import com.kakao.actionbase.v2.core.code.KeyValue
+import com.kakao.actionbase.v2.engine.GraphDefaults
+import com.kakao.actionbase.v2.engine.storage.jdbc.MetadataTable
+
+import org.jetbrains.exposed.sql.Database
+import org.jetbrains.exposed.sql.SortOrder
+import org.jetbrains.exposed.sql.selectAll
+import org.jetbrains.exposed.sql.transactions.transaction
+
+import reactor.core.publisher.Mono
+import reactor.core.scheduler.Schedulers
+
+/**
+ * Reads the JDBC metastore's rows as they sit on disk, tombstones included. No read path above this
+ * one shows them, and a scan key crowded with tombstones truncates silently.
+ */
+class JdbcMetastoreInspector(
+    private val metastore: Database,
+    private val table: MetadataTable,
+) {
+    /** Ordered by `id`, the primary key, so paging walks an index that already exists. */
+    fun dump(
+        limit: Int,
+        offset: Long,
+    ): Mono<List<JdbcMetastoreRow>> =
+        Mono
+            .fromCallable {
+                transaction(metastore) {
+                    table
+                        .selectAll()
+                        .orderBy(table.id, SortOrder.ASC)
+                        .limit(limit, offset)
+                        .map { row ->
+                            JdbcMetastoreRow(
+                                id = row[table.id].value,
+                                k = row[table.k],
+                                decoded = decode(row[table.k], row[table.v]),
+                            )
+                        }
+                }
+            }.subscribeOn(Schedulers.boundedElastic())
+
+    fun count(): Mono<Long> =
+        Mono
+            .fromCallable { transaction(metastore) { table.selectAll().count() } }
+            .subscribeOn(Schedulers.boundedElastic())
+
+    // Null when the codec cannot read the row, which still fills its window and is still reported.
+    // Exception rather than runCatching, which would also swallow an Error and pass a broken JVM off
+    // as one more undecodable row.
+    private fun decode(
+        k: String,
+        v: String,
+    ): JdbcMetastoreEdge? =
+        try {
+            val edge = DecodedEdge.fromMetastore(KeyValue(k, v), emptyMap())
+            JdbcMetastoreEdge(active = edge.isActive, src = edge.src, tgt = edge.tgt, labelId = edge.labelId)
+        } catch (e: Exception) {
+            null
+        }
+
+    companion object {
+        // From the graph, not its parts: the server module cannot name Exposed's Database.
+        fun of(graph: GraphDefaults): JdbcMetastoreInspector = JdbcMetastoreInspector(graph.metastore, graph.metadataTable)
+    }
+}
+
+/** `v` is not carried: it is the largest column and nothing reads it. */
+data class JdbcMetastoreRow(
+    val id: Long,
+    val k: String,
+    val decoded: JdbcMetastoreEdge?,
+)
+
+/**
+ * `src` and `labelId` are the scan key `DdlService.getAll` builds, so they name the window whose
+ * limit is at stake; `active` says whether the row is a tombstone. Nothing else is needed. `src` and
+ * `tgt` stay `Any?` because the codec types them as `Object`.
+ */
+data class JdbcMetastoreEdge(
+    val active: Boolean,
+    val src: Any?,
+    val tgt: Any?,
+    val labelId: Int,
+)

engine/src/test/kotlin/com/kakao/actionbase/v2/engine/metastore/JdbcMetastoreInspectorTest.kt

@@ -0,0 +1,110 @@
+package com.kakao.actionbase.v2.engine.metastore
+
+import com.kakao.actionbase.v2.engine.Graph
+import com.kakao.actionbase.v2.engine.entity.EntityName
+import com.kakao.actionbase.v2.engine.service.ddl.ServiceCreateRequest
+import com.kakao.actionbase.v2.engine.service.ddl.ServiceDeleteRequest
+import com.kakao.actionbase.v2.engine.service.ddl.ServiceUpdateRequest
+import com.kakao.actionbase.v2.engine.storage.jdbc.BaseTableConstants
+import com.kakao.actionbase.v2.engine.test.GraphFixtures
+
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+import org.jetbrains.exposed.sql.insert
+import org.jetbrains.exposed.sql.transactions.transaction
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.TestInstance
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class JdbcMetastoreInspectorTest {
+    private lateinit var graph: Graph
+    private lateinit var inspector: JdbcMetastoreInspector
+
+    @BeforeEach
+    fun setup() {
+        // withTestData: a bare fixture graph keeps its metadata in the local store, leaving the
+        // JDBC table empty.
+        graph = GraphFixtures.create(withTestData = true)
+        inspector = JdbcMetastoreInspector(graph.metastore, graph.metadataTable)
+    }
+
+    @AfterEach
+    fun teardown() {
+        graph.close()
+    }
+
+    private fun dump(
+        limit: Int,
+        offset: Long,
+    ): List<JdbcMetastoreRow> = inspector.dump(limit, offset).block()!!
+
+    @Test
+    fun `counts every row the metastore holds`() {
+        val count = inspector.count().block()!!
+        val all = dump(count.toInt() + 10, 0)
+
+        assertTrue(count > 0, "the fixture writes metadata to the metastore")
+        assertEquals(count, all.size.toLong(), "the count and the rows must agree")
+    }
+
+    @Test
+    fun `pages without dropping or repeating a row`() {
+        val all = dump(1000, 0)
+        val paged = (0 until all.size step 2).flatMap { dump(2, it.toLong()) }
+
+        assertEquals(all.map { it.id }, paged.map { it.id }, "paging must cover the table exactly once")
+    }
+
+    @Test
+    fun `orders by id ascending, whatever the caller asks`() {
+        val ids = dump(1000, 0).map { it.id }
+
+        assertEquals(ids.sorted(), ids, "id ASC is the only order this serves")
+    }
+
+    @Test
+    fun `decodes a row into the window it belongs to`() {
+        val decoded = dump(1000, 0).mapNotNull { it.decoded }
+
+        assertTrue(decoded.isNotEmpty(), "metadata rows decode")
+        assertTrue(decoded.all { it.src != null }, "src names the scan window and must survive the decode")
+        assertTrue(decoded.any { it.active }, "a bootstrapped graph has live metadata")
+    }
+
+    @Test
+    fun `keeps a row the codec cannot read, so it still occupies its window`() {
+        transaction(graph.metastore) {
+            graph.metadataTable.insert {
+                it[k] = "not-an-encoded-key"
+                it[v] = "not-an-encoded-value"
+                it[createdBy] = javaClass.canonicalName.take(BaseTableConstants.MAX_LENGTH)
+                it[modifiedBy] = javaClass.canonicalName.take(BaseTableConstants.MAX_LENGTH)
+            }
+        }
+
+        val undecodable = dump(1000, 0).single { it.k == "not-an-encoded-key" }
+
+        assertNull(undecodable.decoded, "the codec cannot read it")
+        assertNotNull(undecodable.k, "but the row is still reported, because it still fills a window")
+    }
+
+    /** The premise: DELETE rewrites the row with `Active.INACTIVE` rather than removing it. */
+    @Test
+    fun `reports a deleted row as a tombstone still occupying its window`() {
+        val name = EntityName.fromOrigin("doomed")
+        graph.serviceDdl.create(name, ServiceCreateRequest(desc = "doomed")).block()
+        val before = inspector.count().block()!!
+
+        graph.serviceDdl.update(name, ServiceUpdateRequest(active = false, desc = null)).block()
+        graph.serviceDdl.delete(name, ServiceDeleteRequest()).block()
+
+        val row = dump(1000, 0).single { it.decoded?.tgt == "doomed" }
+        assertEquals(false, row.decoded?.active, "a deleted row is a tombstone, not a removal")
+        assertEquals(before, inspector.count().block()!!, "and it still takes up its place in the table")
+    }
+}

server/src/main/kotlin/com/kakao/actionbase/server/api/graph/v2/metastore/JdbcMetastoreController.kt

@@ -0,0 +1,35 @@
+package com.kakao.actionbase.server.api.graph.v2.metastore
+
+import com.kakao.actionbase.v2.engine.metastore.JdbcMetastoreInspector
+import com.kakao.actionbase.v2.engine.metastore.JdbcMetastoreRow
+
+import org.springframework.data.domain.Page
+import org.springframework.data.domain.PageImpl
+import org.springframework.data.domain.Pageable
+import org.springframework.data.web.PageableDefault
+import org.springframework.web.bind.annotation.GetMapping
+import org.springframework.web.bind.annotation.RestController
+
+import reactor.core.publisher.Mono
+
+/**
+ * Pages the JDBC metastore's rows, tombstones included, so `jdbc-metastore remaining` can measure how
+ * much room each scan window has left.
+ *
+ * Narrower than the endpoint #418 removed: no `prefix` (it built `k LIKE '<prefix>%'` unescaped over
+ * base64url keys, whose `_` is a LIKE wildcard), no `sort`, no `/local`. The request shape still
+ * matches 0.4.x so one client reads both.
+ */
+@RestController
+class JdbcMetastoreController(
+    private val metastoreInspector: JdbcMetastoreInspector,
+) {
+    @GetMapping("/graph/v2/metastore/global")
+    fun metastoreGlobal(
+        @PageableDefault(size = 100, page = 0) pageable: Pageable,
+    ): Mono<Page<JdbcMetastoreRow>> =
+        metastoreInspector
+            .dump(pageable.pageSize, pageable.offset)
+            .zipWith(metastoreInspector.count())
+            .map { PageImpl(it.t1, pageable, it.t2) }
+}

server/src/main/kotlin/com/kakao/actionbase/server/configuration/GraphConfiguration.kt

@@ -12,6 +12,7 @@ import com.kakao.actionbase.v2.engine.cdc.CdcFactory
 import com.kakao.actionbase.v2.engine.cdc.DefaultCdcFactory
 import com.kakao.actionbase.v2.engine.client.kafka.KafkaClientFactory
 import com.kakao.actionbase.v2.engine.client.web.WebClientFactory
+import com.kakao.actionbase.v2.engine.metastore.JdbcMetastoreInspector
 import com.kakao.actionbase.v2.engine.util.getLogger
 import com.kakao.actionbase.v2.engine.v3.V2BackedEngine
 import com.kakao.actionbase.v2.engine.wal.DefaultWalFactory
@@ -133,6 +134,9 @@ class GraphConfiguration {
         return graph
     }
 
+    @Bean
+    fun provideJdbcMetastoreInspector(graph: Graph): JdbcMetastoreInspector = JdbcMetastoreInspector.of(graph)
+
     @Bean
     fun provideV2BackedEngine(graph: Graph): V2BackedEngine = V2BackedEngine(graph)
 

server/src/test/kotlin/com/kakao/actionbase/server/filter/ReadOnlyRequestFilterTest.kt

@@ -148,6 +148,7 @@ class ReadOnlyRequestFilterTest {
                 "GET /graph/v2/admin/metadata/service/{service}/label",
                 "GET /graph/v2/admin/metadata/storage",
                 "GET /graph/v2/admin//migration/{name}",
+                "GET /graph/v2/metastore/global",
                 "GET /graph/v2/service",
                 "GET /graph/v2/service/{service}",
                 "GET /graph/v2/service/{service}/alias",