From 38d772c250c611afac20ea35a74beaaf2eefbd77 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 16 Jun 2026 12:57:54 +0100 Subject: [PATCH] Add Kotlin catalog transport helpers --- docs/cultnet-transport-parity.md | 5 + docs/runtime-parity-scope.md | 2 +- packages/cultmesh-kotlin/README.md | 27 +- .../kotlin/org/gamecult/cultmesh/CultMesh.kt | 265 +++++++++++++++++- 4 files changed, 281 insertions(+), 18 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index d785cc29..5841b188 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -309,6 +309,11 @@ Current progress: `rudp://host:port` endpoints and can build a RUDP client directly from a peer card contact hint. The facade delegates to the same owners as the top-level helpers instead of creating a second authority path. +- Kotlin WebSocket and RUDP transports now expose catalog fetch/sync helpers for + schema and shard catalogs. The helpers send standard schema-v0 requests, + require the matching response schema, and apply through the caller's catalog, + giving JVM/Android callers the same client ergonomics Python already had + without making the transport own catalog truth. - Kotlin authority lease catalogs now also expose watch callbacks, so JVM and Android clients can react to lease changes without treating peer-card contact hints as trust. diff --git a/docs/runtime-parity-scope.md b/docs/runtime-parity-scope.md index b5e3c6d0..12502592 100644 --- a/docs/runtime-parity-scope.md +++ b/docs/runtime-parity-scope.md @@ -34,7 +34,7 @@ CultMesh. | Python | Package runtime, local daemon endpoint, raw interop peer, typed state participant | CultCache/CultNet/CultMesh package surface, schema catalogs, shard catalogs/logs, local TCP CultMesh server, session/prediction/simulation fact helpers, authority leases, streams, RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, benchmark evidence against C# cache paths | Universal C# throughput, full C# daemon/server stack, and RUDP as the only public service pipe | | TypeScript | Node/browser-adjacent projection and interop runtime | CultCache parity tests, CultNet schema-v0 peer, transport profiles, TCP-framed transport port and peer helper, shared RUDP socket transport, RUDP reconnect policy/controller, RUDP reconnect-loop helper, local CultMesh node/catalog/authority/stream ergonomics | Full C# game session/server stack | | Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, small CultMesh facade for peer cards, authority leases, RUDP endpoint parsing, and peer-card RUDP client construction | Full CultMesh game-session/server facade, stream catalog, or production game-session server | -| Kotlin | JVM/Android client substrate | Typed MessagePack cache, schema/shard catalogs, authority leases, stream catalog, branded `CultMesh` facade, WebSocket transport profile/connection helper, RUDP client/server sugar, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence | C# SoA store, full C# server daemon, or Unity hot-loop memory layout | +| Kotlin | JVM/Android client substrate | Typed MessagePack cache, schema/shard catalogs with transport fetch/sync helpers, authority leases, stream catalog, branded `CultMesh` facade, WebSocket transport profile/connection helper, RUDP client/server sugar, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence | C# SoA store, full C# server daemon, or Unity hot-loop memory layout | ## Evidence Gates diff --git a/packages/cultmesh-kotlin/README.md b/packages/cultmesh-kotlin/README.md index 801a82da..f3204d61 100644 --- a/packages/cultmesh-kotlin/README.md +++ b/packages/cultmesh-kotlin/README.md @@ -87,13 +87,24 @@ check(frame?.channelId == "schema") transport.sendSchemaMessage(cultNetSchemaCatalogRequest(messageId = "schemas")) val message = transport.receiveSchemaMessage() check(message?.schemaVersion == "cultnet.schema_catalog_response.v0") + +// Or let the transport perform the standard request/response/apply cycle. +val schemas = CultMesh.createSchemaCatalog() +transport.syncSchemaCatalog( + schemas, + messageId = "schemas", + includeSchemaJson = true, + kinds = listOf("document_payload"), +) ``` The WebSocket adapter advertises a `websocket` transport profile with one reliable ordered `schema` channel and exposes transfer stats. `sendSchema`, `sendSchemaMessage`, `receiveSchema`, and `receiveSchemaMessage` keep the -schema lane at the same ergonomic level as RUDP. It is the stream adapter; RUDP -remains the portable realtime UDP path. +schema lane at the same ergonomic level as RUDP. `fetchSchemaCatalog`, +`syncSchemaCatalog`, `fetchShardCatalog`, and `syncShardCatalog` mirror the +Python client helpers without creating a second catalog owner. It is the stream +adapter; RUDP remains the portable realtime UDP path. Schema catalogs are also first-class, so Kotlin peers can publish and consume descriptor responses instead of hand-assembling maps: @@ -187,6 +198,13 @@ CultMesh.createRudpServer( check(pumpRudpPairUntilConnected(client, server)) client.sendSchema("client-state") val payload = server.receiveSchema(timeoutMs = 1_000) + + val remoteSchemas = CultMesh.createSchemaCatalog() + client.syncSchemaCatalog( + remoteSchemas, + messageId = "rudp-schemas", + kinds = listOf("document_payload"), + ) } } ``` @@ -197,7 +215,10 @@ harness. `sendSchema`, `sendLatest`, and `sendRealtime` select the shared channel semantics; they do not create a Kotlin-only dialect. For a remote peer that already has its own receive loop, use `connectAndWait(...)`; for two same-process transports, use `pumpRudpPairUntilConnected(...)` to drive both -sides through the handshake. +sides through the handshake. RUDP transports expose the same catalog helpers as +WebSocket, with timeout-aware receive loops: `fetchSchemaCatalog`, +`fetchSchemaDescriptors`, `syncSchemaCatalog`, `fetchShardCatalog`, +`fetchShardDescriptors`, and `syncShardCatalog`. For lower-level code, `CultMesh.parseRudpEndpoint("rudp://127.0.0.1:4100")` returns the host/port pair, and `CultMesh.createRudpClient(...)` accepts either diff --git a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt index daf64293..75a1c610 100644 --- a/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt +++ b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt @@ -2214,9 +2214,13 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { nextSequenceAfterFrame: Long, expectedSequenceIfUninitialized: Long, ): List { - val next = orderedNextSequenceByChannel[frame.channelId] ?: minOf(expectedSequenceIfUninitialized, frame.sequence).also { + var next = orderedNextSequenceByChannel[frame.channelId] ?: minOf(expectedSequenceIfUninitialized, frame.sequence).also { orderedNextSequenceByChannel[frame.channelId] = it } + while (frame.sequence > next && receivedSequences.contains(next) && orderedBuffers[frame.channelId]?.containsKey(next) != true) { + next += 1 + orderedNextSequenceByChannel[frame.channelId] = next + } if (frame.sequence < next) return emptyList() if (frame.sequence > next) { orderedBuffers.getOrPut(frame.channelId) { TreeMap() }[frame.sequence] = PendingOrderedFrame(frame, nextSequenceAfterFrame) @@ -2234,10 +2238,19 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) { val pending = buffer.remove(next) ?: break delivered.add(pending.frame) orderedNextSequenceByChannel[channelId] = pending.nextSequence + skipReceivedNonChannelSequences(channelId) } return delivered } + private fun skipReceivedNonChannelSequences(channelId: String) { + var next = orderedNextSequenceByChannel[channelId] ?: return + while (receivedSequences.contains(next) && orderedBuffers[channelId]?.containsKey(next) != true) { + next += 1 + orderedNextSequenceByChannel[channelId] = next + } + } + private fun allocateFragmentId(): Int { val fragmentId = nextFragmentId nextFragmentId += 1 @@ -2920,14 +2933,19 @@ private fun cultNetWebSocketTransportCarriesSchemaFramesWithStats() { output.flush() val request = parseCultNetMessage(readMaskedWebSocketBinaryPayload(input)) check(request.schemaVersion == "cultnet.schema_catalog_request.v0") + val catalog = CultNetSchemaCatalog() + catalog.upsert( + defineCultNetSchemaDescriptor( + schemaId = "kotlin.websocket.note.v1", + kind = "document_payload", + documentType = "kotlin.websocket.note", + title = "Kotlin WebSocket Note", + schemaJson = """{"type":"string"}""", + ), + ) writeUnmaskedWebSocketBinaryPayload( output, - cultNetHello( - runtimeId = "kotlin-websocket-server", - runtimeKind = "kotlin", - displayName = "Kotlin WebSocket Server", - supportedMessageVersions = listOf("cultnet.hello.v0"), - ).toBytes(), + catalog.createResponse(request).toBytes(), ) } } catch (error: Throwable) { @@ -2945,11 +2963,15 @@ private fun cultNetWebSocketTransportCarriesSchemaFramesWithStats() { check(descriptor.protocol == "websocket") check(descriptor.channels == listOf(CultNetTransportChannel("schema", "reliable", "ordered"))) - transport.sendSchemaMessage(cultNetSchemaCatalogRequest(messageId = "websocket-schema")) - val response = transport.receiveSchemaMessage() - ?: error("WebSocket transport did not receive schema message") - check(response.schemaVersion == "cultnet.hello.v0") - check(response.body["runtimeId"] == "kotlin-websocket-server") + val synced = CultNetSchemaCatalog() + val descriptors = transport.syncSchemaCatalog( + synced, + messageId = "websocket-schema", + includeSchemaJson = true, + kinds = listOf("document_payload"), + ) + check(descriptors.single().schemaId == "kotlin.websocket.note.v1") + check(synced.get("kotlin.websocket.note.v1")?.documentType == "kotlin.websocket.note") check(transport.stats.framesSent == 1L) check(transport.stats.framesReceived == 1L) } @@ -3627,7 +3649,7 @@ private fun rudpSocketTransportErgonomicFactoriesCarrySchemaFrames() { CultMesh.createRudpServer( runtimeId = "kotlin-rudp-sugar-server", connectionId = connectionId, - tuning = CultNetRudpSocketTuning(resendDelayMs = 25, maxFragmentBytes = 8, maxPendingReliablePackets = 16), + tuning = CultNetRudpSocketTuning(resendDelayMs = 25, maxFragmentBytes = 1024, maxPendingReliablePackets = 16), ).use { server -> val endpoint = CultMesh.parseRudpEndpoint("rudp://127.0.0.1:${server.localPort}") check(endpoint.host == "127.0.0.1") @@ -3642,14 +3664,91 @@ private fun rudpSocketTransportErgonomicFactoriesCarrySchemaFrames() { runtimeId = "kotlin-rudp-sugar-client", connectionId = connectionId, peer = peer, - tuning = CultNetRudpSocketTuning(resendDelayMs = 25, maxFragmentBytes = 8, maxPendingReliablePackets = 16), + tuning = CultNetRudpSocketTuning(resendDelayMs = 25, maxFragmentBytes = 1024, maxPendingReliablePackets = 16), ).use { client -> client.connect("join") check(pumpRudpPairUntilConnected(client, server)) + val schemaCatalog = CultNetSchemaCatalog() + schemaCatalog.upsert( + defineCultNetSchemaDescriptor( + schemaId = "kotlin.rudp.note.v1", + kind = "document_payload", + documentType = "kotlin.rudp.note", + title = "Kotlin RUDP Note", + schemaJson = """{"type":"string"}""", + ), + ) + val syncedSchemas = CultNetSchemaCatalog() + var schemaDescriptors: List? = null + var clientError: Throwable? = null + val schemaThread = Thread { + try { + schemaDescriptors = client.syncSchemaCatalog( + syncedSchemas, + messageId = "rudp-schema", + includeSchemaJson = true, + kinds = listOf("document_payload"), + timeoutMs = 2_000, + ) + } catch (error: Throwable) { + clientError = error + } + } + schemaThread.isDaemon = true + schemaThread.start() + val schemaRequest = server.receiveSchemaMessage(2_000) + ?: error("RUDP server did not receive schema catalog request") + server.sendSchemaMessage(schemaCatalog.createResponse(schemaRequest)) + pollRudpAfterSend(server, 50) + schemaThread.join(2_000) + if (schemaThread.isAlive) error("RUDP schema helper client did not finish") + clientError?.let { throw it } + check(schemaDescriptors?.single()?.schemaId == "kotlin.rudp.note.v1") + check(syncedSchemas.get("kotlin.rudp.note.v1")?.documentType == "kotlin.rudp.note") + server.receiveOnce() + + val shardCatalog = CultNetShardCatalog() + shardCatalog.upsert( + CultNetShardDescriptor( + shardId = "kotlin-rudp-shard", + ownerRuntimeId = "kotlin-rudp-sugar-server", + epoch = 3, + isPrimary = true, + schemaIds = listOf("kotlin.rudp.note.v1"), + keyPrefix = "note:", + ), + ) + var shardDescriptors: List? = null + clientError = null + val shardThread = Thread { + try { + shardDescriptors = client.fetchShardDescriptors( + messageId = "rudp-shards", + schemaIds = listOf("kotlin.rudp.note.v1"), + recordKeys = listOf("note:1"), + timeoutMs = 2_000, + ) + } catch (error: Throwable) { + clientError = error + } + } + shardThread.isDaemon = true + shardThread.start() + val shardRequest = server.receiveSchemaMessage(2_000) + ?: error("RUDP server did not receive shard catalog request") + server.sendSchemaMessage(shardCatalog.createResponse(shardRequest)) + pollRudpAfterSend(server, 50) + shardThread.join(2_000) + if (shardThread.isAlive) error("RUDP shard helper client did not finish") + clientError?.let { throw it } + check(shardDescriptors?.single()?.shardId == "kotlin-rudp-shard") + client.sendSchema("client-state") check(String(server.receiveSchema(1_000) ?: error("Server did not receive schema frame"), StandardCharsets.UTF_8) == "client-state") + client.receiveOnce() server.sendSchema("server-state") check(String(client.receiveSchema(1_000) ?: error("Client did not receive schema frame"), StandardCharsets.UTF_8) == "server-state") + server.receiveOnce() } } } @@ -3987,6 +4086,144 @@ class CultNetWebSocketTransportConnection( } } +fun CultNetWebSocketTransportConnection.fetchSchemaCatalog( + messageId: String = "kotlin-schema-catalog", + includeSchemaJson: Boolean = false, + schemaIds: List = emptyList(), + kinds: List = emptyList(), +): CultNetMessage { + sendSchemaMessage(cultNetSchemaCatalogRequest(messageId, includeSchemaJson, schemaIds, kinds)) + return requireSchemaResponse( + receiveSchemaMessage(), + "cultnet.schema_catalog_response.v0", + "WebSocket schema catalog", + ) +} + +fun CultNetWebSocketTransportConnection.fetchSchemaDescriptors( + messageId: String = "kotlin-schema-catalog", + includeSchemaJson: Boolean = false, + schemaIds: List = emptyList(), + kinds: List = emptyList(), +): List = + CultNetSchemaCatalog().applyResponse(fetchSchemaCatalog(messageId, includeSchemaJson, schemaIds, kinds)) + +fun CultNetWebSocketTransportConnection.syncSchemaCatalog( + catalog: CultNetSchemaCatalog, + messageId: String = "kotlin-schema-catalog", + includeSchemaJson: Boolean = false, + schemaIds: List = emptyList(), + kinds: List = emptyList(), +): List = + catalog.applyResponse(fetchSchemaCatalog(messageId, includeSchemaJson, schemaIds, kinds)) + +fun CultNetWebSocketTransportConnection.fetchShardCatalog( + messageId: String = "kotlin-shard-catalog", + schemaIds: List = emptyList(), + recordKeys: List = emptyList(), +): CultNetMessage { + sendSchemaMessage(cultNetShardCatalogRequest(messageId, schemaIds, recordKeys)) + return requireSchemaResponse( + receiveSchemaMessage(), + "cultnet.shard_catalog_response.v0", + "WebSocket shard catalog", + ) +} + +fun CultNetWebSocketTransportConnection.fetchShardDescriptors( + messageId: String = "kotlin-shard-catalog", + schemaIds: List = emptyList(), + recordKeys: List = emptyList(), +): List = + CultNetShardCatalog().applyResponse(fetchShardCatalog(messageId, schemaIds, recordKeys)) + +fun CultNetWebSocketTransportConnection.syncShardCatalog( + catalog: CultNetShardCatalog, + messageId: String = "kotlin-shard-catalog", + schemaIds: List = emptyList(), + recordKeys: List = emptyList(), +): List = + catalog.applyResponse(fetchShardCatalog(messageId, schemaIds, recordKeys)) + +fun CultNetRudpSocketTransportConnection.fetchSchemaCatalog( + messageId: String = "kotlin-schema-catalog", + includeSchemaJson: Boolean = false, + schemaIds: List = emptyList(), + kinds: List = emptyList(), + timeoutMs: Long = 1_000, + pollIntervalMs: Long = 5, +): CultNetMessage { + sendSchemaMessage(cultNetSchemaCatalogRequest(messageId, includeSchemaJson, schemaIds, kinds)) + return requireSchemaResponse( + receiveSchemaMessage(timeoutMs, pollIntervalMs), + "cultnet.schema_catalog_response.v0", + "RUDP schema catalog", + ) +} + +fun CultNetRudpSocketTransportConnection.fetchSchemaDescriptors( + messageId: String = "kotlin-schema-catalog", + includeSchemaJson: Boolean = false, + schemaIds: List = emptyList(), + kinds: List = emptyList(), + timeoutMs: Long = 1_000, + pollIntervalMs: Long = 5, +): List = + CultNetSchemaCatalog().applyResponse(fetchSchemaCatalog(messageId, includeSchemaJson, schemaIds, kinds, timeoutMs, pollIntervalMs)) + +fun CultNetRudpSocketTransportConnection.syncSchemaCatalog( + catalog: CultNetSchemaCatalog, + messageId: String = "kotlin-schema-catalog", + includeSchemaJson: Boolean = false, + schemaIds: List = emptyList(), + kinds: List = emptyList(), + timeoutMs: Long = 1_000, + pollIntervalMs: Long = 5, +): List = + catalog.applyResponse(fetchSchemaCatalog(messageId, includeSchemaJson, schemaIds, kinds, timeoutMs, pollIntervalMs)) + +fun CultNetRudpSocketTransportConnection.fetchShardCatalog( + messageId: String = "kotlin-shard-catalog", + schemaIds: List = emptyList(), + recordKeys: List = emptyList(), + timeoutMs: Long = 1_000, + pollIntervalMs: Long = 5, +): CultNetMessage { + sendSchemaMessage(cultNetShardCatalogRequest(messageId, schemaIds, recordKeys)) + return requireSchemaResponse( + receiveSchemaMessage(timeoutMs, pollIntervalMs), + "cultnet.shard_catalog_response.v0", + "RUDP shard catalog", + ) +} + +fun CultNetRudpSocketTransportConnection.fetchShardDescriptors( + messageId: String = "kotlin-shard-catalog", + schemaIds: List = emptyList(), + recordKeys: List = emptyList(), + timeoutMs: Long = 1_000, + pollIntervalMs: Long = 5, +): List = + CultNetShardCatalog().applyResponse(fetchShardCatalog(messageId, schemaIds, recordKeys, timeoutMs, pollIntervalMs)) + +fun CultNetRudpSocketTransportConnection.syncShardCatalog( + catalog: CultNetShardCatalog, + messageId: String = "kotlin-shard-catalog", + schemaIds: List = emptyList(), + recordKeys: List = emptyList(), + timeoutMs: Long = 1_000, + pollIntervalMs: Long = 5, +): List = + catalog.applyResponse(fetchShardCatalog(messageId, schemaIds, recordKeys, timeoutMs, pollIntervalMs)) + +private fun requireSchemaResponse(message: CultNetMessage?, expectedSchemaVersion: String, label: String): CultNetMessage { + val response = message ?: throw IOException("$label request did not receive a schema message") + if (response.schemaVersion != expectedSchemaVersion) { + throw IOException("Expected $expectedSchemaVersion, received ${response.schemaVersion}") + } + return response +} + class MessagePackReader(payload: ByteArray) { private val input = DataInputStream(ByteArrayInputStream(payload)) private var pushed = -1