GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 2 Projects Releases Packages Wiki Activity Actions 3

Add Kotlin catalog transport helpers

Browse source
This commit is contained in:
Meta Vens 2026-06-16 12:57:54 +01:00
commit 38d772c250
4 changed files with 281 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

5
docs/cultnet-transport-parity.md
View file

@ -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.

2
docs/runtime-parity-scope.md
View file

@ -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

27
packages/cultmesh-kotlin/README.md
View file

@ -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

265
packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt
View file

@ -2214,9 +2214,13 @@ class CultNetRudpSession(options: CultNetRudpSessionOptions) {
nextSequenceAfterFrame: Long,
expectedSequenceIfUninitialized: Long,
): List<CultNetRudpDeliveredFrame> {
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<CultNetSchemaDescriptor>? = 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<CultNetShardDescriptor>? = 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<String> = emptyList(),
kinds: List<String> = 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<String> = emptyList(),
kinds: List<String> = emptyList(),
): List<CultNetSchemaDescriptor> =
CultNetSchemaCatalog().applyResponse(fetchSchemaCatalog(messageId, includeSchemaJson, schemaIds, kinds))
fun CultNetWebSocketTransportConnection.syncSchemaCatalog(
catalog: CultNetSchemaCatalog,
messageId: String = "kotlin-schema-catalog",
includeSchemaJson: Boolean = false,
schemaIds: List<String> = emptyList(),
kinds: List<String> = emptyList(),
): List<CultNetSchemaDescriptor> =
catalog.applyResponse(fetchSchemaCatalog(messageId, includeSchemaJson, schemaIds, kinds))
fun CultNetWebSocketTransportConnection.fetchShardCatalog(
messageId: String = "kotlin-shard-catalog",
schemaIds: List<String> = emptyList(),
recordKeys: List<String> = 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<String> = emptyList(),
recordKeys: List<String> = emptyList(),
): List<CultNetShardDescriptor> =
CultNetShardCatalog().applyResponse(fetchShardCatalog(messageId, schemaIds, recordKeys))
fun CultNetWebSocketTransportConnection.syncShardCatalog(
catalog: CultNetShardCatalog,
messageId: String = "kotlin-shard-catalog",
schemaIds: List<String> = emptyList(),
recordKeys: List<String> = emptyList(),
): List<CultNetShardDescriptor> =
catalog.applyResponse(fetchShardCatalog(messageId, schemaIds, recordKeys))
fun CultNetRudpSocketTransportConnection.fetchSchemaCatalog(
messageId: String = "kotlin-schema-catalog",
includeSchemaJson: Boolean = false,
schemaIds: List<String> = emptyList(),
kinds: List<String> = 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<String> = emptyList(),
kinds: List<String> = emptyList(),
timeoutMs: Long = 1_000,
pollIntervalMs: Long = 5,
): List<CultNetSchemaDescriptor> =
CultNetSchemaCatalog().applyResponse(fetchSchemaCatalog(messageId, includeSchemaJson, schemaIds, kinds, timeoutMs, pollIntervalMs))
fun CultNetRudpSocketTransportConnection.syncSchemaCatalog(
catalog: CultNetSchemaCatalog,
messageId: String = "kotlin-schema-catalog",
includeSchemaJson: Boolean = false,
schemaIds: List<String> = emptyList(),
kinds: List<String> = emptyList(),
timeoutMs: Long = 1_000,
pollIntervalMs: Long = 5,
): List<CultNetSchemaDescriptor> =
catalog.applyResponse(fetchSchemaCatalog(messageId, includeSchemaJson, schemaIds, kinds, timeoutMs, pollIntervalMs))
fun CultNetRudpSocketTransportConnection.fetchShardCatalog(
messageId: String = "kotlin-shard-catalog",
schemaIds: List<String> = emptyList(),
recordKeys: List<String> = 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<String> = emptyList(),
recordKeys: List<String> = emptyList(),
timeoutMs: Long = 1_000,
pollIntervalMs: Long = 5,
): List<CultNetShardDescriptor> =
CultNetShardCatalog().applyResponse(fetchShardCatalog(messageId, schemaIds, recordKeys, timeoutMs, pollIntervalMs))
fun CultNetRudpSocketTransportConnection.syncShardCatalog(
catalog: CultNetShardCatalog,
messageId: String = "kotlin-shard-catalog",
schemaIds: List<String> = emptyList(),
recordKeys: List<String> = emptyList(),
timeoutMs: Long = 1_000,
pollIntervalMs: Long = 5,
): List<CultNetShardDescriptor> =
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