diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 5913ae93..cc3b2a6c 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -32,7 +32,7 @@ CultNet owns cross-runtime transport semantics: | TypeScript `cultnet-ts` | `CultNetPeer` over any Node `Duplex`, TCP-framed transport, or single-peer RUDP socket transport; interop uses TCP | UDP multicast probe/announce in the interop peer | First UDP socket binding for the shared RUDP reliability owner | | Rust `cultnet-rs` | Interop example serves and dials schema-v0 MessagePack over TCP-framed or shared RUDP transport; single-peer RUDP socket transport exists in the library | UDP multicast probe/announce with TCP and RUDP transport profiles | UDP socket binding for the shared RUDP reliability owner | | Python `cultcache-py` | TCP sockets with 4-byte length-prefixed MessagePack frames for local CultMesh/CultNet server and client; single-peer RUDP socket transport exists in the library | Endpoint lists and CultMesh peer/Verse catalogs | UDP socket binding for the shared RUDP reliability owner | -| Kotlin `cultmesh-kotlin` | Channel-aware WebSocket transport connection for reliable ordered `schema` frames; single-peer RUDP socket transport exists in the library | CultMesh Verse and peer catalogs, plus endpoint lists carried in schema-v0 catalog messages | WebSocket adapter plus UDP socket binding for the shared RUDP reliability owner, with build-script self-test | +| Kotlin `cultmesh-kotlin` | Channel-aware WebSocket transport connection for reliable ordered `schema` frames; single-peer RUDP socket transport exists in the library; interop CLI serves and dials schema-v0 MessagePack over TCP-framed or shared RUDP transport | CultMesh Verse and peer catalogs, plus endpoint lists carried in schema-v0 catalog messages | WebSocket/TCP adapters plus UDP socket binding for the shared RUDP reliability owner, with build-script and TypeScript harness proof | The live split is therefore: payload language and the native RUDP packet/session language now converge across the targeted runtimes, while production service @@ -360,6 +360,12 @@ Current progress: the socket transport. The TypeScript interop peer now also serves and dials the schema-v0 interop flow over `interop-rudp` while keeping TCP-framed as the compatibility lane; discovery and hello responses advertise both profiles. +- Kotlin now has the same full interop RUDP proof shape as the other targeted + runtimes: the JVM CLI can serve schema-v0 hello/catalog/snapshot/mutation/fire + over `interop-rudp`, and it can dial a TypeScript full peer over the same + shared RUDP connection id and reliable ordered `schema` channel. TCP-framed + remains a compatibility lane for the interop peer; WebSocket remains the + JVM/Android-friendly client transport surface. - TypeScript now also has a `CultNetSchemaCatalog` for remote schema descriptors plus `CultNetPeer` request/fetch/sync helpers for `cultnet.schema_catalog_request.v0` / `cultnet.schema_catalog_response.v0`. diff --git a/docs/runtime-parity-scope.md b/docs/runtime-parity-scope.md index bbb68d48..57485a4e 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, branded CultMesh node/schema/shard/Verse/peer/authority/stream/RUDP ergonomics, authorized peer lookup, shard catalogs/logs, schema-transport factory for raw client request/subscription flows, local TCP and RUDP CultMesh server listeners, Python interop RUDP serve/dial path, session/prediction/simulation fact helpers, 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, advertised schema/shard catalog wire contracts plus request/sync helpers, transport profiles, TCP-framed transport port and peer helper, shared RUDP socket transport, RUDP interop serve/dial path, RUDP reconnect policy/controller, RUDP reconnect-loop helper, branded CultMesh node/schema/shard/Verse/peer/authority/stream/RUDP ergonomics, authorized peer lookup | Full C# game session/server stack | | Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, schema and shard catalog helpers, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport, Rust interop RUDP serve/dial path, connected RUDP CultMesh client helpers, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, small CultMesh facade for schema registry, shard catalog, peer cards, authority leases, authorized peer lookup, RUDP endpoint parsing, and peer-card/authorized-peer RUDP client construction | Full CultMesh game-session/server facade, stream catalog, or production game-session server | -| Kotlin | JVM/Android client substrate | Typed MessagePack cache, built-in schema catalog advertisement, schema/shard catalogs with shared schema-message transport fetch/sync helpers, authority leases, authorized peer lookup, stream catalog, branded `CultMesh` facade, WebSocket transport profile/connection helper, RUDP client/server sugar, connected RUDP CultMesh client helpers, 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, built-in schema catalog advertisement, schema/shard catalogs with shared schema-message transport fetch/sync helpers, authority leases, authorized peer lookup, stream catalog, branded `CultMesh` facade, WebSocket transport profile/connection helper, RUDP client/server sugar, connected RUDP CultMesh client helpers, Kotlin interop RUDP serve/dial path, 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/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt b/packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt index 39a2ca5f..4b7bd68a 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 @@ -3,7 +3,9 @@ package org.gamecult.cultmesh import java.io.ByteArrayInputStream import java.io.ByteArrayOutputStream import java.io.DataInputStream +import java.io.DataOutputStream import java.io.EOFException +import java.io.File import java.io.IOException import java.io.InputStream import java.io.OutputStream @@ -2953,6 +2955,8 @@ fun main(args: Array) { "rudp-dial-once" -> rudpDialOnce(options) "rudp-serve-message-once" -> rudpServeSchemaMessageOnce(options) "rudp-dial-message-once" -> rudpDialSchemaMessageOnce(options) + "serve" -> interopServe(options) + "dial" -> interopDial(options) else -> error("Unknown mode ${args[0]}") } } @@ -3194,6 +3198,785 @@ private fun cultNetReconnectControllerSchedulesAttemptsAndReset() { check(controller.canAttempt(99_000)) } +private const val interopDocumentType = "cultnet.interop-note" +private const val interopSchemaVersion = "cultnet.interop_note.v0" +private const val interopMutationIntentType = "cultnet.interop-note-mutation-intent" +private const val interopMutationIntentSchemaId = "https://github.com/GameCult/cultnet-ts/integration/contracts/cultnet.interop-note-mutation-intent.schema.json" +private const val interopMutationIntentSchemaVersion = "cultnet.interop_note_mutation_intent.v0" +private const val interopMutationReceiptType = "cultnet.interop-note-mutation-receipt" +private const val interopMutationReceiptSchemaId = "https://github.com/GameCult/cultnet-ts/integration/contracts/cultnet.interop-note-mutation-receipt.schema.json" +private const val interopMutationReceiptSchemaVersion = "cultnet.interop_note_mutation_receipt.v0" +private const val interopFireCommandType = "cultnet.interop-fire-weapon-command" +private const val interopFireCommandSchemaId = "https://github.com/GameCult/cultnet-ts/integration/contracts/cultnet.interop-fire-weapon-command.schema.json" +private const val interopFireCommandSchemaVersion = "cultnet.interop_fire_weapon_command.v0" +private const val interopFireReceiptType = "cultnet.interop-fire-weapon-receipt" +private const val interopFireReceiptSchemaId = "https://github.com/GameCult/cultnet-ts/integration/contracts/cultnet.interop-fire-weapon-receipt.schema.json" +private const val interopFireReceiptSchemaVersion = "cultnet.interop_fire_weapon_receipt.v0" +private const val rudpInteropConnectionId = 0x43554c54L +private const val rudpInteropMaxFragmentBytes = 1024 +private const val rudpInteropResendDelayMs = 25L +private val interopReadTimeoutMs = 15_000L + +private data class InteropServeConfig( + val runtimeId: String, + val runtimeKind: String, + val displayName: String, + val agentId: String, + val bindHost: String, + val advertiseHost: String, + val tcpPort: Int, + val rudpPort: Int, + val discoveryPort: Int, + val discoveryGroup: String, + val schemaPath: String, +) + +private data class InteropDialConfig( + val runtimeId: String, + val runtimeKind: String, + val displayName: String, + val agentId: String, + val targetHost: String, + val targetPort: Int?, + val targetRudpPort: Int?, + val schemaPath: String, +) + +private data class InteropSchemaRegistration(val schemaId: String, val title: String?, val schemaJson: String) +private data class InteropDocuments( + val note: CultDocumentDefinition, + val mutationIntent: CultDocumentDefinition, + val mutationReceipt: CultDocumentDefinition, + val fireCommand: CultDocumentDefinition, + val fireReceipt: CultDocumentDefinition, +) + +private data class InteropNote( + val schemaVersion: String = interopSchemaVersion, + val documentId: String, + val authorRuntimeId: String, + val title: String, + val body: String, + val tags: List, +) + +private data class InteropMutationIntent( + val schemaVersion: String = interopMutationIntentSchemaVersion, + val intentId: String, + val targetDocumentId: String, + val appendBody: String, + val appendTag: String, +) + +private data class InteropMutationReceipt( + val schemaVersion: String = interopMutationReceiptSchemaVersion, + val intentId: String, + val accepted: Boolean, + val documentId: String, + val body: String, + val tags: List, +) + +private data class InteropFireCommand( + val schemaVersion: String = interopFireCommandSchemaVersion, + val commandId: String, + val characterId: String, + val weaponId: String, +) + +private data class InteropFireReceipt( + val schemaVersion: String = interopFireReceiptSchemaVersion, + val commandId: String, + val accepted: Boolean, + val characterId: String, + val weaponId: String, + val shotsFired: Long, + val ammoRemaining: Long, +) + +private class InteropCodec( + override val documentType: String, + override val schemaVersion: String, + private val encodeValue: (T) -> List, + private val decodeValue: (List) -> T, +) : CultDocumentCodec { + override fun encode(value: T): ByteArray = MessagePackWriter().value(encodeValue(value)).toByteArray() + override fun decode(payload: ByteArray): T = decodeValue(anyList(MessagePackReader(payload).readAny())) +} + +private fun interopServe(options: Map) { + val config = parseInteropServeConfig(options) + val registration = loadInteropSchemaRegistration(config.schemaPath) + val documents = defineInteropDocuments(registration.schemaId) + val cache = CultCache() + registerInteropDocuments(cache, documents) + val schemaCatalog = CultNetSchemaCatalog() + schemaCatalog.upsert( + defineCultNetSchemaDescriptor( + schemaId = registration.schemaId, + kind = "document_payload", + schemaVersion = interopSchemaVersion, + documentType = interopDocumentType, + title = registration.title, + wireContracts = listOf("cultnet.schema.v0"), + schemaJson = registration.schemaJson, + ), + ) + cache.put(documents.note, "note:${config.runtimeId}", buildInteropNote(config.runtimeId, config.displayName)) + + startInteropTcpServer(config, cache, documents, schemaCatalog) + startInteropRudpServer(config, cache, documents, schemaCatalog) + + println(jsonLine(linkedMapOf( + "status" to "ready", + "mode" to "serve", + "runtimeId" to config.runtimeId, + "runtimeKind" to config.runtimeKind, + "tcpPort" to config.tcpPort, + "rudpPort" to config.rudpPort, + "discoveryPort" to config.discoveryPort, + "discoveryGroup" to config.discoveryGroup, + ))) + + while (true) Thread.sleep(3_600_000) +} + +private fun interopDial(options: Map) { + val config = parseInteropDialConfig(options) + val registration = loadInteropSchemaRegistration(config.schemaPath) + val documents = defineInteropDocuments(registration.schemaId) + val cache = CultCache() + registerInteropDocuments(cache, documents) + val transportName = if (config.targetRudpPort != null) "rudp" else "tcp_framed" + val transport = openInteropDialTransport(config) + transport.use { + it.sendSchemaMessage( + cultNetHello( + runtimeId = config.runtimeId, + runtimeKind = config.runtimeKind, + displayName = config.displayName, + supportedDocumentTypes = listOf(interopDocumentType), + supportedMutationContracts = listOf(interopMutationContract()), + supportedMessageVersions = listOf(interopSchemaVersion), + transportProfiles = dialInteropTransportProfiles(config), + supportsSchemaCatalog = true, + ), + ) + val remoteHello = requireSchemaResponse(it.receiveSchemaMessage(interopReadTimeoutMs, 5), "cultnet.hello.v0", "interop hello") + val remoteRuntimeId = requireWireString(remoteHello.body, "runtimeId") + + it.sendSchemaMessage( + CultNetMessage( + "cultnet.schema_catalog_request.v0", + linkedMapOf("messageId" to "${config.runtimeId}-catalog", "includeSchemaJson" to true), + ), + ) + val catalogResponse = requireSchemaResponse(it.receiveSchemaMessage(interopReadTimeoutMs, 5), "cultnet.schema_catalog_response.v0", "interop schema catalog") + val hasInteropSchema = mapList(catalogResponse.body["schemas"]).any { + it["schemaId"] == registration.schemaId && it["documentType"] == interopDocumentType + } + + it.sendSchemaMessage(CultNetMessage("cultnet.snapshot_request.v0", linkedMapOf("messageId" to "${config.runtimeId}-snapshot"))) + val snapshotResponse = requireSchemaResponse(it.receiveSchemaMessage(interopReadTimeoutMs, 5), "cultnet.snapshot_response_raw.v0", "interop snapshot") + val snapshotDocuments = mapList(snapshotResponse.body["documents"]).map { rawDocumentRecordFromWire(it) } + val decodedSnapshotNotes = snapshotDocuments + .mapNotNull { document -> runCatching { documents.note.codec.decode(document.payload) }.getOrNull() } + val note = cache.applyRawSnapshotResponse(snapshotResponse) + .filterIsInstance() + .firstOrNull { candidate -> candidate.authorRuntimeId == remoteRuntimeId } + ?: decodedSnapshotNotes.firstOrNull { candidate -> candidate.authorRuntimeId == remoteRuntimeId } + ?: throw IOException( + "did not receive an interop note from $remoteRuntimeId; snapshotDocuments=${snapshotDocuments.size}; " + + "schemaIds=${snapshotDocuments.map { document -> document.schemaId }}; " + + "decodedAuthors=${decodedSnapshotNotes.map { decoded -> decoded.authorRuntimeId }}", + ) + + val mutationIntent = InteropMutationIntent( + intentId = "${config.runtimeId}-decorate", + targetDocumentId = note.documentId, + appendBody = " Decorated by ${config.runtimeId}.", + appendTag = "decorated:${config.runtimeId}", + ) + it.sendSchemaMessage(rawPut("${config.runtimeId}-decorate-put", interopMutationIntentSchemaId, mutationIntent.intentId, documents.mutationIntent.codec.encode(mutationIntent))) + val mutationReceiptMessage = readDocumentPutFor(it, interopMutationReceiptSchemaId) + val mutationReceipt = cache.applyRawDocumentPut(mutationReceiptMessage) as InteropMutationReceipt + val mutatedNoteMessage = readDocumentPutFor(it, registration.schemaId) + val mutatedNote = cache.applyRawDocumentPut(mutatedNoteMessage) as InteropNote + + val fireCommand = InteropFireCommand( + commandId = "${config.runtimeId}-fire", + characterId = remoteRuntimeId, + weaponId = "interop-rifle", + ) + it.sendSchemaMessage(rawPut("${config.runtimeId}-fire-put", interopFireCommandSchemaId, fireCommand.commandId, documents.fireCommand.codec.encode(fireCommand))) + val fireReceipt = cache.applyRawDocumentPut(readDocumentPutFor(it, interopFireReceiptSchemaId)) as InteropFireReceipt + + println(jsonLine(linkedMapOf( + "mode" to "dial", + "runtimeId" to config.runtimeId, + "targetHost" to config.targetHost, + "targetPort" to (config.targetRudpPort ?: config.targetPort), + "transport" to transportName, + "remoteHello" to linkedMapOf("schemaVersion" to "cultnet.hello.v0", "runtimeId" to remoteRuntimeId), + "hasInteropSchema" to hasInteropSchema, + "retrievedNote" to interopNoteWire(note), + "mutatedNote" to interopNoteWire(mutatedNote), + "mutationReceipt" to interopMutationReceiptWire(mutationReceipt), + "fireReceipt" to interopFireReceiptWire(fireReceipt), + ))) + } +} + +private fun startInteropTcpServer( + config: InteropServeConfig, + cache: CultCache, + documents: InteropDocuments, + schemaCatalog: CultNetSchemaCatalog, +) { + val server = ServerSocket(config.tcpPort, 50, InetAddress.getByName(config.bindHost)) + val thread = Thread { + while (true) { + try { + val socket = server.accept() + Thread { + try { + InteropTcpFramedTransport( + socket, + createTcpFramedInteropProfile(config.runtimeId, config.advertiseHost, config.tcpPort), + ).use { transport -> + while (true) { + val message = transport.receiveSchemaMessage(interopReadTimeoutMs, 5) ?: break + handleInteropServerMessage(transport, message, config, cache, documents, schemaCatalog) + } + } + } catch (error: Throwable) { + System.err.println("kotlin interop tcp error: ${error.message}") + } + }.also { it.isDaemon = true; it.start() } + } catch (error: Throwable) { + System.err.println("kotlin interop tcp accept error: ${error.message}") + } + } + } + thread.isDaemon = true + thread.start() +} + +private fun startInteropRudpServer( + config: InteropServeConfig, + cache: CultCache, + documents: InteropDocuments, + schemaCatalog: CultNetSchemaCatalog, +) { + val socket = DatagramSocket(config.rudpPort, InetAddress.getByName(config.bindHost)).also { it.soTimeout = 20 } + val sessions = linkedMapOf() + val thread = Thread { + val buffer = ByteArray(65535) + while (true) { + try { + val datagram = DatagramPacket(buffer, buffer.size) + socket.receive(datagram) + val remote = InetSocketAddress(datagram.address, datagram.port) + val packet = try { + decodeRudpPacket(buffer.copyOf(datagram.length)) + } catch (_: IOException) { + continue + } + if (packet.connectionId != rudpInteropConnectionId) continue + val session = sessions.getOrPut(remote) { + CultNetRudpSession(CultNetRudpSessionOptions(rudpInteropConnectionId, 100, rudpInteropResendDelayMs)) + } + if (packet.packetType == CultNetRudpPacketType.Connect) { + sendInteropRudpPacket(socket, remote, session.acceptConnect(packet, nowMs(), "cultnet-interop-rudp".toByteArray(StandardCharsets.UTF_8))) + continue + } + val result = session.receive(packet, nowMs()) + result.reply?.let { sendInteropRudpPacket(socket, remote, it) } + if (packet.packetType == CultNetRudpPacketType.Data) sendInteropRudpPacket(socket, remote, session.createAck()) + for (frame in result.delivered) { + if (frame.channelId != "schema") continue + val sender = InteropRudpSessionSender(socket, remote, session) + handleInteropServerMessage(sender, parseCultNetMessage(frame.payload), config, cache, documents, schemaCatalog) + } + } catch (_: SocketTimeoutException) { + val now = nowMs() + for ((remote, session) in sessions) { + session.dueResends(now).forEach { sendInteropRudpPacket(socket, remote, it) } + } + } catch (error: Throwable) { + System.err.println("kotlin interop rudp error: ${error.message}") + } + } + } + thread.isDaemon = true + thread.start() +} + +private fun handleInteropServerMessage( + transport: CultNetSchemaMessageTransport, + message: CultNetMessage, + config: InteropServeConfig, + cache: CultCache, + documents: InteropDocuments, + schemaCatalog: CultNetSchemaCatalog, +) { + synchronized(cache) { + when (message.schemaVersion) { + "cultnet.hello.v0" -> transport.sendSchemaMessage( + cultNetHello( + runtimeId = config.runtimeId, + runtimeKind = config.runtimeKind, + displayName = config.displayName, + supportedDocumentTypes = listOf(interopDocumentType), + supportedMutationContracts = listOf(interopMutationContract()), + supportedMessageVersions = listOf(interopSchemaVersion), + transportProfiles = listOf( + createTcpFramedInteropProfile(config.runtimeId, config.advertiseHost, config.tcpPort), + createRudpTransportProfile(config.runtimeId, transportId = "interop-rudp", host = config.advertiseHost, port = config.rudpPort, maxFragmentBytes = rudpInteropMaxFragmentBytes), + ), + supportsSchemaCatalog = true, + ), + ) + "cultnet.schema_catalog_request.v0" -> transport.sendSchemaMessage(schemaCatalog.createResponse(message)) + "cultnet.snapshot_request.v0" -> { + val response = cache.createRawSnapshotResponse(message, sourceRuntimeId = config.runtimeId) + val documentsWithSource = mapList(response.body["documents"]).map { + rawDocumentRecordFromWire(it).copy( + sourceRuntimeId = config.runtimeId, + sourceAgentId = config.agentId, + sourceRole = "peer", + tags = listOf("interop", config.runtimeId), + ) + } + transport.sendSchemaMessage(cultNetSnapshotResponseRaw(response.body["messageId"] as? String ?: "", documentsWithSource)) + } + "cultnet.document_put_raw.v0" -> handleInteropRawPut(transport, message, config, cache, documents) + } + } +} + +private fun handleInteropRawPut( + transport: CultNetSchemaMessageTransport, + message: CultNetMessage, + config: InteropServeConfig, + cache: CultCache, + documents: InteropDocuments, +) { + val document = rawDocumentRecordFromWire(mapValue(message.body["document"])) + when (document.schemaId) { + interopMutationIntentSchemaId -> { + val intent = cache.applyRawDocumentPut(message) as InteropMutationIntent + val note = cache.getRequired(documents.note, intent.targetDocumentId) + val mutated = note.copy( + body = note.body + intent.appendBody, + tags = note.tags + intent.appendTag, + ) + cache.put(documents.note, mutated.documentId, mutated) + val receipt = InteropMutationReceipt( + intentId = intent.intentId, + accepted = true, + documentId = mutated.documentId, + body = mutated.body, + tags = mutated.tags, + ) + transport.sendSchemaMessage(rawPut("${config.runtimeId}-mutation-receipt", interopMutationReceiptSchemaId, receipt.intentId, documents.mutationReceipt.codec.encode(receipt), config, "mutation")) + transport.sendSchemaMessage(rawPut("${config.runtimeId}-mutated-note", documents.note.schemaVersion, mutated.documentId, documents.note.codec.encode(mutated), config, "mutation")) + } + interopFireCommandSchemaId -> { + val command = cache.applyRawDocumentPut(message) as InteropFireCommand + val receipt = InteropFireReceipt( + commandId = command.commandId, + accepted = true, + characterId = command.characterId, + weaponId = command.weaponId, + shotsFired = 1, + ammoRemaining = 29, + ) + transport.sendSchemaMessage(rawPut("${config.runtimeId}-fire-receipt", interopFireReceiptSchemaId, receipt.commandId, documents.fireReceipt.codec.encode(receipt), config, "side-effect")) + } + } +} + +private class InteropTcpFramedTransport( + private val socket: Socket, + override val profile: CultNetTransportProfile, +) : CultNetSchemaMessageTransport { + private val input = DataInputStream(socket.getInputStream()) + private val output = DataOutputStream(socket.getOutputStream()) + override var stats: CultNetTransportStats = CultNetTransportStats() + private set + + override fun sendSchemaMessage(message: CultNetMessage) { + val payload = message.toBytes() + synchronized(output) { + output.writeInt(payload.size) + output.write(payload) + output.flush() + } + stats = stats.copy(bytesSent = stats.bytesSent + payload.size + 4, framesSent = stats.framesSent + 1) + } + + override fun receiveSchemaMessage(timeoutMs: Long, pollIntervalMs: Long): CultNetMessage? { + val previousTimeout = socket.soTimeout + socket.soTimeout = timeoutMs.coerceAtLeast(1).coerceAtMost(Int.MAX_VALUE.toLong()).toInt() + return try { + val length = input.readInt() + val payload = ByteArray(length) + input.readFully(payload) + stats = stats.copy(bytesReceived = stats.bytesReceived + payload.size + 4, framesReceived = stats.framesReceived + 1) + parseCultNetMessage(payload) + } catch (_: SocketTimeoutException) { + null + } catch (_: EOFException) { + null + } finally { + socket.soTimeout = previousTimeout + } + } + + override fun close() { + socket.close() + } +} + +private class InteropRudpSessionSender( + private val socket: DatagramSocket, + private val remote: InetSocketAddress, + private val session: CultNetRudpSession, +) : CultNetSchemaMessageTransport { + override val profile: CultNetTransportProfile = createRudpTransportProfile("kotlin-interop-rudp-session", transportId = "interop-rudp") + override val stats: CultNetTransportStats = CultNetTransportStats() + + override fun sendSchemaMessage(message: CultNetMessage) { + session.sendMany("schema", message.toBytes(), CultNetRudpSendOptions(reliable = true, ordered = true, nowMs = nowMs()), rudpInteropMaxFragmentBytes) + .forEach { sendInteropRudpPacket(socket, remote, it) } + } + + override fun receiveSchemaMessage(timeoutMs: Long, pollIntervalMs: Long): CultNetMessage? = null + override fun close() {} +} + +private fun openInteropDialTransport(config: InteropDialConfig): CultNetSchemaMessageTransport { + val rudpPort = config.targetRudpPort + if (rudpPort != null) { + val transport = cultNetRudpClient( + runtimeId = "${config.runtimeId}-interop-rudp-dial", + connectionId = rudpInteropConnectionId, + remoteHost = config.targetHost, + remotePort = rudpPort, + tuning = CultNetRudpSocketTuning(resendDelayMs = rudpInteropResendDelayMs, maxFragmentBytes = rudpInteropMaxFragmentBytes), + ) + if (!transport.connectAndWait("cultnet-interop-rudp", timeoutMs = 5_000, pollIntervalMs = 5)) { + throw IOException("timed out waiting for RUDP interop connect") + } + return transport + } + val tcpPort = config.targetPort ?: throw IOException("dial requires --target-port or --target-rudp-port") + val socket = Socket(config.targetHost, tcpPort) + return InteropTcpFramedTransport(socket, createTcpFramedInteropProfile(config.runtimeId, config.targetHost, tcpPort)) +} + +private fun createTcpFramedInteropProfile(runtimeId: String, host: String, port: Int): CultNetTransportProfile = + CultNetTransportProfile( + runtimeId = runtimeId, + transports = listOf( + CultNetTransportDescriptor( + transportId = "interop-tcp", + protocol = "tcp_framed", + wireContracts = listOf("cultnet.schema.v0"), + host = host, + port = port, + channels = listOf(CultNetTransportChannel("schema", "reliable", "ordered")), + ), + ), + ) + +private fun dialInteropTransportProfiles(config: InteropDialConfig): List { + val profiles = mutableListOf() + if (config.targetPort != null) profiles.add(createTcpFramedInteropProfile(config.runtimeId, config.targetHost, config.targetPort)) + if (config.targetRudpPort != null) profiles.add(createRudpTransportProfile(config.runtimeId, transportId = "interop-rudp", host = config.targetHost, port = config.targetRudpPort, maxFragmentBytes = rudpInteropMaxFragmentBytes)) + return profiles +} + +private fun sendInteropRudpPacket(socket: DatagramSocket, remote: InetSocketAddress, packet: CultNetRudpPacket) { + val wire = encodeRudpPacket(packet) + socket.send(DatagramPacket(wire, wire.size, remote.address, remote.port)) +} + +private fun readDocumentPutFor(transport: CultNetSchemaMessageTransport, schemaId: String): CultNetMessage { + val deadline = nowMs() + interopReadTimeoutMs + while (nowMs() < deadline) { + val message = transport.receiveSchemaMessage(250, 5) ?: continue + if (message.schemaVersion == "cultnet.document_put_raw.v0") { + val document = mapValue(message.body["document"]) + if (document["schemaId"] == schemaId) return message + } + } + throw IOException("timed out waiting for document put $schemaId") +} + +private fun rawPut( + messageId: String, + schemaId: String, + recordKey: String, + payload: ByteArray, + config: InteropServeConfig? = null, + tag: String? = null, +): CultNetMessage = cultNetDocumentPutRaw( + messageId = messageId, + document = CultNetRawDocumentRecord( + schemaId = schemaId, + recordKey = recordKey, + storedAt = Instant.now().toString(), + payload = payload, + sourceRuntimeId = config?.runtimeId, + sourceAgentId = config?.agentId, + sourceRole = config?.let { "peer" }, + tags = listOfNotNull(tag, config?.runtimeId), + ), +) + +private fun defineInteropDocuments(noteSchemaId: String): InteropDocuments { + val note = CultDocumentDefinition(InteropCodec(interopDocumentType, noteSchemaId, ::interopNoteSlots, ::interopNoteFromSlots)) + val mutationIntent = CultDocumentDefinition(InteropCodec(interopMutationIntentType, interopMutationIntentSchemaId, ::interopMutationIntentSlots, ::interopMutationIntentFromSlots)) + val mutationReceipt = CultDocumentDefinition(InteropCodec(interopMutationReceiptType, interopMutationReceiptSchemaId, ::interopMutationReceiptSlots, ::interopMutationReceiptFromSlots)) + val fireCommand = CultDocumentDefinition(InteropCodec(interopFireCommandType, interopFireCommandSchemaId, ::interopFireCommandSlots, ::interopFireCommandFromSlots)) + val fireReceipt = CultDocumentDefinition(InteropCodec(interopFireReceiptType, interopFireReceiptSchemaId, ::interopFireReceiptSlots, ::interopFireReceiptFromSlots)) + return InteropDocuments(note, mutationIntent, mutationReceipt, fireCommand, fireReceipt) +} + +private fun registerInteropDocuments(cache: CultCache, documents: InteropDocuments) { + cache.register(documents.note) + cache.register(documents.mutationIntent) + cache.register(documents.mutationReceipt) + cache.register(documents.fireCommand) + cache.register(documents.fireReceipt) +} + +private fun buildInteropNote(runtimeId: String, displayName: String): InteropNote = InteropNote( + documentId = "note:$runtimeId", + authorRuntimeId = runtimeId, + title = "$displayName keeps a little note", + body = "$runtimeId can move CultNet state without begging the gods for translation.", + tags = listOf(runtimeId, "interop", "cultnet"), +) + +private fun interopMutationContract(): Map = linkedMapOf( + "documentType" to interopDocumentType, + "payloadSchemaVersion" to interopSchemaVersion, + "operations" to listOf("snapshot", "documentPut", "intentSubmit", "receiptWatch"), + "authority" to "runtime", + "intentDocumentTypes" to listOf(interopMutationIntentType, interopFireCommandType), + "receiptDocumentTypes" to listOf(interopMutationReceiptType, interopFireReceiptType), +) + +private fun loadInteropSchemaRegistration(schemaPath: String): InteropSchemaRegistration { + val schemaJson = File(schemaPath).readText(StandardCharsets.UTF_8) + return InteropSchemaRegistration( + schemaId = extractJsonString(schemaJson, "\$id") ?: throw IOException("schema $schemaPath is missing \$id"), + title = extractJsonString(schemaJson, "title"), + schemaJson = schemaJson, + ) +} + +private fun extractJsonString(json: String, field: String): String? { + val pattern = Regex(""""${Regex.escape(field)}"\s*:\s*"((?:\\.|[^"\\])*)"""") + return pattern.find(json)?.groupValues?.get(1)?.replace("\\\"", "\"") +} + +private fun parseInteropServeConfig(options: Map): InteropServeConfig { + val tcpPort = requiredIntArg(options, "tcp-port") + return InteropServeConfig( + runtimeId = requiredArg(options, "runtime-id"), + runtimeKind = requiredArg(options, "runtime-kind"), + displayName = requiredArg(options, "display-name"), + agentId = requiredArg(options, "agent-id"), + bindHost = options["bind-host"] ?: "127.0.0.1", + advertiseHost = requiredArg(options, "advertise-host"), + tcpPort = tcpPort, + rudpPort = options["rudp-port"]?.toInt() ?: tcpPort, + discoveryPort = requiredIntArg(options, "discovery-port"), + discoveryGroup = requiredArg(options, "discovery-group"), + schemaPath = requiredArg(options, "schema-path"), + ) +} + +private fun parseInteropDialConfig(options: Map): InteropDialConfig = InteropDialConfig( + runtimeId = requiredArg(options, "runtime-id"), + runtimeKind = requiredArg(options, "runtime-kind"), + displayName = requiredArg(options, "display-name"), + agentId = requiredArg(options, "agent-id"), + targetHost = requiredArg(options, "target-host"), + targetPort = options["target-port"]?.toInt(), + targetRudpPort = options["target-rudp-port"]?.toInt(), + schemaPath = requiredArg(options, "schema-path"), +).also { + if (it.targetPort == null && it.targetRudpPort == null) throw IOException("dial requires --target-port or --target-rudp-port") +} + +private fun requiredArg(options: Map, name: String): String = + options[name] ?: throw IOException("missing --$name") + +private fun requiredIntArg(options: Map, name: String): Int = + requiredArg(options, name).toInt() + +private fun interopNoteWire(value: InteropNote): Map = linkedMapOf( + "schemaVersion" to value.schemaVersion, + "documentId" to value.documentId, + "authorRuntimeId" to value.authorRuntimeId, + "title" to value.title, + "body" to value.body, + "tags" to value.tags, +) + +private fun interopNoteSlots(value: InteropNote): List = listOf( + value.schemaVersion, + value.documentId, + value.authorRuntimeId, + value.title, + value.body, + value.tags, +) + +private fun interopNoteFromSlots(slots: List): InteropNote = InteropNote( + schemaVersion = slots.stringAt(0, interopSchemaVersion), + documentId = slots.requiredStringAt(1, "documentId"), + authorRuntimeId = slots.requiredStringAt(2, "authorRuntimeId"), + title = slots.requiredStringAt(3, "title"), + body = slots.requiredStringAt(4, "body"), + tags = stringList(slots.getOrNull(5)), +) + +private fun interopMutationIntentWire(value: InteropMutationIntent): Map = linkedMapOf( + "schemaVersion" to value.schemaVersion, + "intentId" to value.intentId, + "targetDocumentId" to value.targetDocumentId, + "appendBody" to value.appendBody, + "appendTag" to value.appendTag, +) + +private fun interopMutationIntentSlots(value: InteropMutationIntent): List = listOf( + value.schemaVersion, + value.intentId, + value.targetDocumentId, + value.appendBody, + value.appendTag, +) + +private fun interopMutationIntentFromSlots(slots: List): InteropMutationIntent = InteropMutationIntent( + schemaVersion = slots.stringAt(0, interopMutationIntentSchemaVersion), + intentId = slots.requiredStringAt(1, "intentId"), + targetDocumentId = slots.requiredStringAt(2, "targetDocumentId"), + appendBody = slots.requiredStringAt(3, "appendBody"), + appendTag = slots.requiredStringAt(4, "appendTag"), +) + +private fun interopMutationReceiptWire(value: InteropMutationReceipt): Map = linkedMapOf( + "schemaVersion" to value.schemaVersion, + "intentId" to value.intentId, + "accepted" to value.accepted, + "documentId" to value.documentId, + "body" to value.body, + "tags" to value.tags, +) + +private fun interopMutationReceiptSlots(value: InteropMutationReceipt): List = listOf( + value.schemaVersion, + value.intentId, + value.accepted, + value.documentId, + value.body, + value.tags, +) + +private fun interopMutationReceiptFromSlots(slots: List): InteropMutationReceipt = InteropMutationReceipt( + schemaVersion = slots.stringAt(0, interopMutationReceiptSchemaVersion), + intentId = slots.requiredStringAt(1, "intentId"), + accepted = slots.getOrNull(2) == true, + documentId = slots.requiredStringAt(3, "documentId"), + body = slots.requiredStringAt(4, "body"), + tags = stringList(slots.getOrNull(5)), +) + +private fun interopFireCommandWire(value: InteropFireCommand): Map = linkedMapOf( + "schemaVersion" to value.schemaVersion, + "commandId" to value.commandId, + "characterId" to value.characterId, + "weaponId" to value.weaponId, +) + +private fun interopFireCommandSlots(value: InteropFireCommand): List = listOf( + value.schemaVersion, + value.commandId, + value.characterId, + value.weaponId, +) + +private fun interopFireCommandFromSlots(slots: List): InteropFireCommand = InteropFireCommand( + schemaVersion = slots.stringAt(0, interopFireCommandSchemaVersion), + commandId = slots.requiredStringAt(1, "commandId"), + characterId = slots.requiredStringAt(2, "characterId"), + weaponId = slots.requiredStringAt(3, "weaponId"), +) + +private fun interopFireReceiptWire(value: InteropFireReceipt): Map = linkedMapOf( + "schemaVersion" to value.schemaVersion, + "commandId" to value.commandId, + "accepted" to value.accepted, + "characterId" to value.characterId, + "weaponId" to value.weaponId, + "shotsFired" to value.shotsFired, + "ammoRemaining" to value.ammoRemaining, +) + +private fun interopFireReceiptSlots(value: InteropFireReceipt): List = listOf( + value.schemaVersion, + value.commandId, + value.accepted, + value.characterId, + value.weaponId, + value.shotsFired, + value.ammoRemaining, +) + +private fun interopFireReceiptFromSlots(slots: List): InteropFireReceipt = InteropFireReceipt( + schemaVersion = slots.stringAt(0, interopFireReceiptSchemaVersion), + commandId = slots.requiredStringAt(1, "commandId"), + accepted = slots.getOrNull(2) == true, + characterId = slots.requiredStringAt(3, "characterId"), + weaponId = slots.requiredStringAt(4, "weaponId"), + shotsFired = (slots.getOrNull(5) as? Number)?.toLong() ?: 0, + ammoRemaining = (slots.getOrNull(6) as? Number)?.toLong() ?: 0, +) + +private fun anyList(value: Any?): List = when (value) { + is Iterable<*> -> value.toList() + is Array<*> -> value.toList() + else -> throw IOException("Expected MessagePack array") +} + +private fun List.requiredStringAt(index: Int, fieldName: String): String { + val value = getOrNull(index) + if (value !is String || value.isBlank()) throw IOException("$fieldName must be a non-empty string") + return value +} + +private fun List.stringAt(index: Int, fallback: String): String = + (getOrNull(index) as? String)?.takeIf { it.isNotBlank() } ?: fallback + +private fun jsonLine(value: Map): String = jsonValue(value) + +private fun jsonValue(value: Any?): String = when (value) { + null -> "null" + is String -> "\"" + value.replace("\\", "\\\\").replace("\"", "\\\"").replace("\n", "\\n").replace("\r", "\\r") + "\"" + is Boolean -> if (value) "true" else "false" + is Number -> value.toString() + is Map<*, *> -> value.entries.joinToString(prefix = "{", postfix = "}") { (key, nested) -> jsonValue(key.toString()) + ":" + jsonValue(nested) } + is Iterable<*> -> value.joinToString(prefix = "[", postfix = "]") { jsonValue(it) } + is Array<*> -> value.joinToString(prefix = "[", postfix = "]") { jsonValue(it) } + else -> jsonValue(value.toString()) +} + private fun cultNetRudpReconnectLoopConsumesSharedController() { var now = 10_000L var capturedTimer: (() -> Unit)? = null diff --git a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts index c43cf290..bec86942 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts @@ -1908,6 +1908,108 @@ test("CultNet TypeScript and C# full interop peers exchange schema-v0 over RUDP" assert.equal(csharpRudpDial.fireReceipt.accepted, true); }); +test("CultNet TypeScript and Kotlin full interop peers exchange schema-v0 over RUDP", async (t) => { + logInteropPhase("ts-kotlin-full-rudp", "build kotlin peer"); + await buildKotlinInteropPeer(); + cleanInteropStores([ + "ts-kotlin-rudp-peer", + "ts-kotlin-rudp-client-dial", + ]); + + const kotlinTcpPort = await getFreePort(); + const kotlinRudpPort = await getFreePort(); + const tsTcpPort = await getFreePort(); + const tsRudpPort = await getFreePort(); + const discoveryPort = await getFreePort(); + const advertiseHost = findAdvertiseHost(); + const servers: RunningServeProcess[] = []; + servers.push(await spawnServeProcess("kotlin-rudp", { + command: kotlinJavaCommand, + args: [ + "-cp", kotlinClasspath, + "org.gamecult.cultmesh.CultMeshKt", + "serve", + "--runtime-id", "kotlin-rudp-peer", + "--runtime-kind", "kotlin", + "--display-name", "Kotlin RUDP Peer", + "--agent-id", "kotlin-rudp-agent", + "--bind-host", "127.0.0.1", + "--advertise-host", advertiseHost, + "--tcp-port", String(kotlinTcpPort), + "--rudp-port", String(kotlinRudpPort), + "--discovery-port", String(discoveryPort), + "--discovery-group", discoveryGroup, + "--schema-path", interopSchemaPath, + ], + cwd: cultLibRoot, + })); + await servers[servers.length - 1].ready; + + servers.push(await spawnServeProcess("ts-kotlin-rudp", { + command: process.execPath, + args: [ + tsPeerScript, + "serve", + "--runtime-id", "ts-kotlin-rudp-peer", + "--runtime-kind", "node", + "--display-name", "TypeScript Kotlin RUDP Peer", + "--agent-id", "ts-kotlin-rudp-agent", + "--bind-host", "127.0.0.1", + "--advertise-host", advertiseHost, + "--tcp-port", String(tsTcpPort), + "--rudp-port", String(tsRudpPort), + "--discovery-port", String(discoveryPort), + "--discovery-group", discoveryGroup, + "--schema-path", interopSchemaPath, + ], + cwd: cultNetTsRoot, + })); + await servers[servers.length - 1].ready; + + t.after(async () => { + await Promise.all(servers.map(stopProcess)); + }); + + logInteropPhase("ts-kotlin-full-rudp", "typescript dials kotlin over rudp"); + const tsRudpDial = await runJsonCommand("ts-kotlin-rudp-dial", process.execPath, [ + tsPeerScript, + "dial", + "--runtime-id", "ts-kotlin-rudp-client", + "--runtime-kind", "node", + "--display-name", "TS Kotlin RUDP Dialer", + "--agent-id", "ts-kotlin-rudp-client-agent", + "--target-host", "127.0.0.1", + "--target-rudp-port", String(kotlinRudpPort), + "--schema-path", interopSchemaPath, + ], cultNetTsRoot); + assert.equal(tsRudpDial.transport, "rudp"); + assert.equal(tsRudpDial.remoteHello.runtimeId, "kotlin-rudp-peer"); + assert.equal(tsRudpDial.hasInteropSchema, true); + assert.equal(tsRudpDial.retrievedNote.authorRuntimeId, "kotlin-rudp-peer"); + assert.ok(tsRudpDial.mutatedNote.tags.includes("decorated:ts-kotlin-rudp-client")); + assert.equal(tsRudpDial.fireReceipt.accepted, true); + + logInteropPhase("ts-kotlin-full-rudp", "kotlin dials typescript over rudp"); + const kotlinRudpDial = await runJsonCommand("kotlin-rudp-dial", kotlinJavaCommand, [ + "-cp", kotlinClasspath, + "org.gamecult.cultmesh.CultMeshKt", + "dial", + "--runtime-id", "kotlin-rudp-client", + "--runtime-kind", "kotlin", + "--display-name", "Kotlin RUDP Dialer", + "--agent-id", "kotlin-rudp-client-agent", + "--target-host", "127.0.0.1", + "--target-rudp-port", String(tsRudpPort), + "--schema-path", interopSchemaPath, + ], cultLibRoot); + assert.equal(kotlinRudpDial.transport, "rudp"); + assert.equal(kotlinRudpDial.remoteHello.runtimeId, "ts-kotlin-rudp-peer"); + assert.equal(kotlinRudpDial.hasInteropSchema, true); + assert.equal(kotlinRudpDial.retrievedNote.authorRuntimeId, "ts-kotlin-rudp-peer"); + assert.ok(kotlinRudpDial.mutatedNote.tags.includes("decorated:kotlin-rudp-client")); + assert.equal(kotlinRudpDial.fireReceipt.accepted, true); +}); + test("CultNet RUDP reliable schema delivery survives one dropped client data packet across runtimes", async () => { for (const peer of rudpServerInteropPeers()) { await peer.build();