Add portable reconnect policy helpers
This commit is contained in:
parent
0569462efd
commit
b3be836115
15 changed files with 412 additions and 12 deletions
|
|
@ -220,6 +220,11 @@ Current progress:
|
|||
connect, accept, single-packet sends, or fragmented sends enqueue anything;
|
||||
TCP-framed profiles do not publish the field because they do not own the RUDP
|
||||
pending-reliable queue.
|
||||
- TypeScript, C#, Rust, Python, and Kotlin now share a portable
|
||||
`cultnet.reconnect_policy.v0` document and deterministic exponential delay
|
||||
helper. This gives RUDP peers the same policy language for reconnect posture;
|
||||
it does not yet claim automatic reconnect orchestration or production service
|
||||
adoption of that policy.
|
||||
- Kotlin now mirrors the C#/TypeScript/Python local authority lease gate:
|
||||
peer cards may advertise `authorityLeaseId`, but `CultMeshAuthorityLeaseCatalog`
|
||||
owns role/shard/time authorization so discovery contact cannot impersonate
|
||||
|
|
@ -239,8 +244,8 @@ Current progress:
|
|||
creating a second authority path.
|
||||
- The remaining parity work is to deepen Python's server-side use of its port
|
||||
and move each runtime's existing TCP/LiteNetLib/WebSocket bodies behind those
|
||||
ports, then broaden RUDP interop into reconnect policy and the wider
|
||||
schema-message matrix.
|
||||
ports, then broaden RUDP interop into automatic reconnect orchestration and
|
||||
the wider schema-message matrix.
|
||||
|
||||
## RUDP Packet Contract V0
|
||||
|
||||
|
|
|
|||
|
|
@ -30,11 +30,11 @@ CultMesh.
|
|||
|
||||
| Runtime | Role | Claimed parity | Not claimed |
|
||||
| --- | --- | --- | --- |
|
||||
| C# | Reference implementation and production game/server substrate | CultCache typed documents and SoA, CultNet schema-v0, LiteNetLib production path, shared transport profiles, shared RUDP socket transport, CultMesh node/session/catalog/stream/simulation surfaces | No need to abandon LiteNetLib before the CultNet adapter fully owns production service adoption |
|
||||
| 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, 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, shared RUDP socket transport, 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, schema-v0 MessagePack over RUDP evidence | Full CultMesh ergonomic facade or production game-session server |
|
||||
| Kotlin | JVM/Android client substrate | Typed MessagePack cache, schema/shard catalogs, authority leases, stream catalog, branded `CultMesh` facade, RUDP client/server sugar, schema-v0 MessagePack over RUDP evidence | C# SoA store, full C# server daemon, or Unity hot-loop memory layout |
|
||||
| C# | Reference implementation and production game/server substrate | CultCache typed documents and SoA, CultNet schema-v0, LiteNetLib production path, shared transport profiles, shared RUDP socket transport, portable reconnect policy helper, CultMesh node/session/catalog/stream/simulation surfaces | No need to abandon LiteNetLib before the CultNet adapter fully owns production service adoption |
|
||||
| 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, portable reconnect policy 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, shared RUDP socket transport, portable reconnect policy 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, portable reconnect policy helper, schema-v0 MessagePack over RUDP evidence | Full CultMesh ergonomic facade or production game-session server |
|
||||
| Kotlin | JVM/Android client substrate | Typed MessagePack cache, schema/shard catalogs, authority leases, stream catalog, branded `CultMesh` facade, RUDP client/server sugar, portable reconnect policy helper, schema-v0 MessagePack over RUDP evidence | C# SoA store, full C# server daemon, or Unity hot-loop memory layout |
|
||||
|
||||
## Evidence Gates
|
||||
|
||||
|
|
@ -43,7 +43,9 @@ CultMesh.
|
|||
TypeScript, Rust, Python, and Kotlin, including loss, reordering,
|
||||
fragmentation, disconnect reason propagation, ping/pong, and schema-v0
|
||||
MessagePack over RUDP for C#, Rust, Python, and Kotlin peers against
|
||||
TypeScript.
|
||||
TypeScript. Runtime-local tests also cover the portable RUDP reconnect policy
|
||||
document and deterministic delay helper in C#, TypeScript, Rust, Python, and
|
||||
Kotlin.
|
||||
- `packages/cultcache-py/PARITY.md` is the Python-specific audit ledger. Its
|
||||
"Still Not Claimed" section is part of the contract, not an apology.
|
||||
- `docs/cultnet-transport-parity.md` is the transport ownership map. It tracks
|
||||
|
|
@ -57,8 +59,10 @@ CultMesh.
|
|||
- Broaden service adoption so local servers and daemon paths use the shared
|
||||
transport port consistently instead of keeping raw TCP/LiteNetLib/WebSocket
|
||||
ownership in individual bodies.
|
||||
- Extend RUDP interop evidence through reconnect policy and the wider
|
||||
schema-message matrix.
|
||||
- Extend RUDP interop evidence through automatic reconnect orchestration and
|
||||
the wider schema-message matrix. The portable reconnect policy document and
|
||||
delay helper exist across the targeted runtimes; service adoption of that
|
||||
policy is a separate claim.
|
||||
- Keep Kotlin ergonomic parity moving with Android/JVM-shaped APIs instead of
|
||||
backfilling C# server responsibilities it should not own.
|
||||
- Keep performance claims evidence-bound: C# owns SoA; Python has benchmark
|
||||
|
|
|
|||
|
|
@ -101,7 +101,8 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti.
|
|||
transport profile advertisement, ping/pong, disconnect reason propagation,
|
||||
timeout state, fragmentation/reassembly, reliable ordered `schema` delivery,
|
||||
unreliable sequenced `latest`, unreliable unordered `realtime`, resend
|
||||
pressure, and bounded reliable-send backpressure. The TypeScript interop
|
||||
pressure, bounded reliable-send backpressure, and the portable
|
||||
`cultnet.reconnect_policy.v0` delay-policy helper. The TypeScript interop
|
||||
harness proves bidirectional Python/TypeScript RUDP socket exchange,
|
||||
packet-loss recovery, reordering recovery, fragmentation, ping/pong, and
|
||||
disconnect reason propagation through the shared wire language.
|
||||
|
|
@ -283,7 +284,9 @@ node --test packages\cultnet-ts\dist-test\test\interop\cultnet-interop.test.js
|
|||
raw client, and interop peer still use the TCP-framed schema lane for many
|
||||
service and harness flows, while `docs/cultnet-transport-parity.md` tracks the
|
||||
remaining work to put broader server-side bodies behind the shared transport
|
||||
port instead of cloning LiteNetLib.
|
||||
port instead of cloning LiteNetLib. The reconnect policy helper is portable;
|
||||
automatic reconnect orchestration is not yet claimed for Python service
|
||||
bodies.
|
||||
- Python does not implement the full C# daemon/server stack. Its current role is
|
||||
package runtime, raw interop peer, launchable local CultMesh endpoint, local
|
||||
CultMesh session facade, and typed state participant.
|
||||
|
|
|
|||
|
|
@ -76,9 +76,12 @@ from .transport import (
|
|||
CultNetRudpSocketMode,
|
||||
CultNetRudpSocketTransportConnection,
|
||||
CultNetRudpSocketTransportOptions,
|
||||
CultNetReconnectPolicy,
|
||||
CultNetTransportFrame,
|
||||
CultNetTransportStats,
|
||||
TcpFramedTransportConnection,
|
||||
compute_reconnect_delay_ms,
|
||||
create_reconnect_policy,
|
||||
create_rudp_transport_profile,
|
||||
create_tcp_framed_transport_profile,
|
||||
decode_rudp_packet,
|
||||
|
|
@ -141,6 +144,7 @@ __all__ = [
|
|||
"CultNetRudpSocketMode",
|
||||
"CultNetRudpSocketTransportConnection",
|
||||
"CultNetRudpSocketTransportOptions",
|
||||
"CultNetReconnectPolicy",
|
||||
"CultNetTransportFrame",
|
||||
"CultNetTransportStats",
|
||||
"TcpFramedTransportConnection",
|
||||
|
|
@ -149,6 +153,8 @@ __all__ = [
|
|||
"apply_raw_snapshot",
|
||||
"apply_shard_log_response",
|
||||
"compute_simulation_claim_hash",
|
||||
"compute_reconnect_delay_ms",
|
||||
"create_reconnect_policy",
|
||||
"create_rudp_transport_profile",
|
||||
"create_tcp_framed_transport_profile",
|
||||
"database_subscribe",
|
||||
|
|
|
|||
|
|
@ -23,6 +23,52 @@ class CultNetTransportStats:
|
|||
frames_sent: int = 0
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CultNetReconnectPolicy:
|
||||
schema_version: str = "cultnet.reconnect_policy.v0"
|
||||
policy_id: str = "default"
|
||||
base_delay_ms: int = 1000
|
||||
max_delay_ms: int = 30000
|
||||
max_jitter_ms: int = 250
|
||||
max_attempts: int | None = None
|
||||
|
||||
def to_wire(self) -> dict[str, Any]:
|
||||
wire: dict[str, Any] = {
|
||||
"schemaVersion": self.schema_version,
|
||||
"policyId": self.policy_id,
|
||||
"baseDelayMs": self.base_delay_ms,
|
||||
"maxDelayMs": self.max_delay_ms,
|
||||
"maxJitterMs": self.max_jitter_ms,
|
||||
}
|
||||
if self.max_attempts is not None:
|
||||
wire["maxAttempts"] = self.max_attempts
|
||||
return wire
|
||||
|
||||
|
||||
def create_reconnect_policy(
|
||||
*,
|
||||
policy_id: str = "default",
|
||||
base_delay_ms: int = 1000,
|
||||
max_delay_ms: int = 30000,
|
||||
max_jitter_ms: int = 250,
|
||||
max_attempts: int | None = None,
|
||||
) -> CultNetReconnectPolicy:
|
||||
return CultNetReconnectPolicy(
|
||||
policy_id=policy_id or "default",
|
||||
base_delay_ms=base_delay_ms,
|
||||
max_delay_ms=max_delay_ms,
|
||||
max_jitter_ms=max_jitter_ms,
|
||||
max_attempts=max_attempts,
|
||||
)
|
||||
|
||||
|
||||
def compute_reconnect_delay_ms(policy: CultNetReconnectPolicy, attempt: int, jitter_ms: int = 0) -> int:
|
||||
normalized_attempt = max(1, int(attempt))
|
||||
capped_base_delay = min(policy.max_delay_ms, policy.base_delay_ms * (2 ** (normalized_attempt - 1)))
|
||||
bounded_jitter = max(0, min(policy.max_jitter_ms, int(jitter_ms)))
|
||||
return capped_base_delay + bounded_jitter
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CultNetTransportFrame:
|
||||
channel_id: str
|
||||
|
|
|
|||
|
|
@ -82,6 +82,7 @@ from cultnet_py import (
|
|||
apply_shard_log_response,
|
||||
create_rudp_transport_profile,
|
||||
create_tcp_framed_transport_profile,
|
||||
create_reconnect_policy,
|
||||
database_subscribe,
|
||||
database_unsubscribe,
|
||||
decode_frame,
|
||||
|
|
@ -107,6 +108,7 @@ from cultnet_py import (
|
|||
wire_message_schema_catalog,
|
||||
wire_message_schema_descriptors,
|
||||
witness_artifact_bundle,
|
||||
compute_reconnect_delay_ms,
|
||||
)
|
||||
from cultmesh_py import create_node
|
||||
from cultmesh_py import (
|
||||
|
|
@ -683,6 +685,19 @@ class CultCacheTests(unittest.TestCase):
|
|||
],
|
||||
)
|
||||
|
||||
def test_cultnet_reconnect_policy_exposes_portable_delay_contract(self) -> None:
|
||||
policy = create_reconnect_policy(policy_id="rudp-default", max_attempts=8)
|
||||
|
||||
self.assertEqual(policy.schema_version, "cultnet.reconnect_policy.v0")
|
||||
self.assertEqual(policy.policy_id, "rudp-default")
|
||||
self.assertEqual(policy.max_attempts, 8)
|
||||
self.assertEqual(policy.to_wire()["policyId"], "rudp-default")
|
||||
self.assertEqual(policy.to_wire()["maxAttempts"], 8)
|
||||
self.assertEqual(compute_reconnect_delay_ms(policy, 1), 1_000)
|
||||
self.assertEqual(compute_reconnect_delay_ms(policy, 3, 17), 4_017)
|
||||
self.assertEqual(compute_reconnect_delay_ms(policy, 9, 999), 30_250)
|
||||
self.assertEqual(compute_reconnect_delay_ms(policy, 0, -5), 1_000)
|
||||
|
||||
def test_cultnet_rudp_session_handshake_acks_reliable_connect_and_accept_packets(self) -> None:
|
||||
client = CultNetRudpSession(
|
||||
CultNetRudpSessionOptions(connection_id=0x0A0B0C0D, initial_sequence=1, resend_delay_ms=50)
|
||||
|
|
|
|||
|
|
@ -167,6 +167,18 @@ that already has its own receive loop, use `connectAndWait(...)`; for two
|
|||
same-process transports, use `pumpRudpPairUntilConnected(...)` to drive both
|
||||
sides through the handshake.
|
||||
|
||||
Kotlin also exposes the shared reconnect-policy helper for peers that need a
|
||||
portable backoff document without inventing a JVM-only delay dialect:
|
||||
|
||||
```kotlin
|
||||
val policy = createReconnectPolicy(policyId = "rudp-default", maxAttempts = 8)
|
||||
val nextDelayMs = computeReconnectDelayMs(policy, attempt = 3, jitterMs = 17)
|
||||
```
|
||||
|
||||
The helper publishes `cultnet.reconnect_policy.v0` and matches the C#,
|
||||
TypeScript, Rust, and Python defaults. It is policy language, not an automatic
|
||||
reconnect loop.
|
||||
|
||||
`EveDashboardStateDocument` mirrors the CultUI-shaped dashboard surface contract:
|
||||
`surface.root` is the retained UI tree, `surface.assets` carries cacheable media
|
||||
references, and the flat `nodes` projection remains a compatibility and binding
|
||||
|
|
|
|||
|
|
@ -325,6 +325,57 @@ data class CultNetTransportStats(
|
|||
|
||||
data class CultNetTransportFrame(val channelId: String, val payload: ByteArray)
|
||||
|
||||
data class CultNetReconnectPolicy(
|
||||
val schemaVersion: String = "cultnet.reconnect_policy.v0",
|
||||
val policyId: String = "default",
|
||||
val baseDelayMs: Long = 1_000,
|
||||
val maxDelayMs: Long = 30_000,
|
||||
val maxJitterMs: Long = 250,
|
||||
val maxAttempts: Int? = null,
|
||||
) {
|
||||
fun toWireMap(): Map<String, Any?> {
|
||||
val wire = linkedMapOf<String, Any?>(
|
||||
"schemaVersion" to schemaVersion,
|
||||
"policyId" to policyId,
|
||||
"baseDelayMs" to baseDelayMs,
|
||||
"maxDelayMs" to maxDelayMs,
|
||||
"maxJitterMs" to maxJitterMs,
|
||||
)
|
||||
if (maxAttempts != null) wire["maxAttempts"] = maxAttempts
|
||||
return wire
|
||||
}
|
||||
}
|
||||
|
||||
fun createReconnectPolicy(
|
||||
policyId: String = "default",
|
||||
baseDelayMs: Long = 1_000,
|
||||
maxDelayMs: Long = 30_000,
|
||||
maxJitterMs: Long = 250,
|
||||
maxAttempts: Int? = null,
|
||||
): CultNetReconnectPolicy = CultNetReconnectPolicy(
|
||||
policyId = policyId.ifBlank { "default" },
|
||||
baseDelayMs = baseDelayMs,
|
||||
maxDelayMs = maxDelayMs,
|
||||
maxJitterMs = maxJitterMs,
|
||||
maxAttempts = maxAttempts,
|
||||
)
|
||||
|
||||
fun computeReconnectDelayMs(policy: CultNetReconnectPolicy, attempt: Int, jitterMs: Long = 0): Long {
|
||||
val normalizedAttempt = attempt.coerceAtLeast(1)
|
||||
val exponent = (normalizedAttempt - 1).coerceAtMost(62)
|
||||
val multiplier = 1L shl exponent
|
||||
val base = policy.baseDelayMs.coerceAtLeast(0)
|
||||
val cappedBaseDelay = if (base == 0L) {
|
||||
0L
|
||||
} else {
|
||||
val maxMultiplier = Long.MAX_VALUE / base
|
||||
val product = if (multiplier > maxMultiplier) Long.MAX_VALUE else base * multiplier
|
||||
product.coerceAtMost(policy.maxDelayMs)
|
||||
}
|
||||
val boundedJitter = jitterMs.coerceIn(0, policy.maxJitterMs)
|
||||
return cappedBaseDelay + boundedJitter
|
||||
}
|
||||
|
||||
data class CultNetTransportProfile(
|
||||
val schemaVersion: String = "cultnet.transport_profile.v0",
|
||||
val runtimeId: String,
|
||||
|
|
@ -2285,6 +2336,7 @@ fun main(args: Array<String>) {
|
|||
cultMeshFacadeRoutesErgonomicEntrypoints()
|
||||
cultCacheRawSnapshotsRoundTripThroughCultNetMessages()
|
||||
cultNetSchemaMessagesUseMessagePackMaps()
|
||||
cultNetReconnectPolicyExposesPortableDelayContract()
|
||||
cultNetSchemaCatalogsRoundTripDescriptors()
|
||||
cultMeshCatalogsRoundTripDiscoveryMessages()
|
||||
cultMeshAuthorityLeasesGatePeerTrust()
|
||||
|
|
@ -2465,6 +2517,21 @@ private fun cultNetSchemaMessagesUseMessagePackMaps() {
|
|||
check((document["payload"] as ByteArray).contentEquals(payload))
|
||||
}
|
||||
|
||||
private fun cultNetReconnectPolicyExposesPortableDelayContract() {
|
||||
val policy = createReconnectPolicy(policyId = "rudp-default", maxAttempts = 8)
|
||||
val wire = policy.toWireMap()
|
||||
|
||||
check(policy.schemaVersion == "cultnet.reconnect_policy.v0")
|
||||
check(policy.policyId == "rudp-default")
|
||||
check(policy.maxAttempts == 8)
|
||||
check(wire["policyId"] == "rudp-default")
|
||||
check(wire["maxAttempts"] == 8)
|
||||
check(computeReconnectDelayMs(policy, 1) == 1_000L)
|
||||
check(computeReconnectDelayMs(policy, 3, 17) == 4_017L)
|
||||
check(computeReconnectDelayMs(policy, 9, 999) == 30_250L)
|
||||
check(computeReconnectDelayMs(policy, 0, -5) == 1_000L)
|
||||
}
|
||||
|
||||
private fun cultNetSchemaCatalogsRoundTripDescriptors() {
|
||||
val catalog = CultNetSchemaCatalog()
|
||||
var watched: CultNetSchemaDescriptor? = null
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
use anyhow::Result;
|
||||
use anyhow::anyhow;
|
||||
use serde::Deserialize;
|
||||
use serde::Serialize;
|
||||
use std::io::Read;
|
||||
use std::io::Write;
|
||||
|
||||
|
|
@ -26,6 +28,68 @@ pub struct CultNetTransportFrame {
|
|||
pub payload: Vec<u8>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct CultNetReconnectPolicy {
|
||||
pub schema_version: String,
|
||||
pub policy_id: String,
|
||||
pub base_delay_ms: u64,
|
||||
pub max_delay_ms: u64,
|
||||
pub max_jitter_ms: u64,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub max_attempts: Option<u32>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct CultNetReconnectPolicyOptions {
|
||||
pub policy_id: String,
|
||||
pub base_delay_ms: u64,
|
||||
pub max_delay_ms: u64,
|
||||
pub max_jitter_ms: u64,
|
||||
pub max_attempts: Option<u32>,
|
||||
}
|
||||
|
||||
impl Default for CultNetReconnectPolicyOptions {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
policy_id: "default".to_string(),
|
||||
base_delay_ms: 1_000,
|
||||
max_delay_ms: 30_000,
|
||||
max_jitter_ms: 250,
|
||||
max_attempts: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn create_reconnect_policy(options: CultNetReconnectPolicyOptions) -> CultNetReconnectPolicy {
|
||||
CultNetReconnectPolicy {
|
||||
schema_version: "cultnet.reconnect_policy.v0".to_string(),
|
||||
policy_id: if options.policy_id.trim().is_empty() {
|
||||
"default".to_string()
|
||||
} else {
|
||||
options.policy_id
|
||||
},
|
||||
base_delay_ms: options.base_delay_ms,
|
||||
max_delay_ms: options.max_delay_ms,
|
||||
max_jitter_ms: options.max_jitter_ms,
|
||||
max_attempts: options.max_attempts,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn compute_reconnect_delay_ms(
|
||||
policy: &CultNetReconnectPolicy,
|
||||
attempt: u32,
|
||||
jitter_ms: u64,
|
||||
) -> u64 {
|
||||
let normalized_attempt = attempt.max(1);
|
||||
let multiplier = 2_u64.saturating_pow(normalized_attempt.saturating_sub(1));
|
||||
let capped_base_delay = policy
|
||||
.base_delay_ms
|
||||
.saturating_mul(multiplier)
|
||||
.min(policy.max_delay_ms);
|
||||
capped_base_delay + jitter_ms.min(policy.max_jitter_ms)
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default)]
|
||||
pub struct TcpFramedTransportProfileOptions {
|
||||
pub transport_id: Option<String>,
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ use cultnet_rs::CultNetDocumentPutOptions;
|
|||
use cultnet_rs::CultNetDocumentRegistry;
|
||||
use cultnet_rs::CultNetMessage;
|
||||
use cultnet_rs::CultNetMutationAuthority;
|
||||
use cultnet_rs::CultNetReconnectPolicyOptions;
|
||||
use cultnet_rs::CultNetRudpPacket;
|
||||
use cultnet_rs::CultNetRudpPacketType;
|
||||
use cultnet_rs::CultNetRudpSendOptions;
|
||||
|
|
@ -35,6 +36,8 @@ use cultnet_rs::LengthPrefixedMessageFramer;
|
|||
use cultnet_rs::TcpFramedTransportConnection;
|
||||
use cultnet_rs::TcpFramedTransportProfileOptions;
|
||||
use cultnet_rs::builtin_schema_registry;
|
||||
use cultnet_rs::compute_reconnect_delay_ms;
|
||||
use cultnet_rs::create_reconnect_policy;
|
||||
use cultnet_rs::create_rudp_transport_profile;
|
||||
use cultnet_rs::create_tcp_framed_transport_profile;
|
||||
use cultnet_rs::decode_cultnet_message_from_slice;
|
||||
|
|
@ -315,6 +318,23 @@ fn rudp_transport_profile_advertises_state_and_realtime_channels() {
|
|||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reconnect_policy_exposes_portable_delay_contract() {
|
||||
let policy = create_reconnect_policy(CultNetReconnectPolicyOptions {
|
||||
policy_id: "rudp-default".to_string(),
|
||||
max_attempts: Some(8),
|
||||
..CultNetReconnectPolicyOptions::default()
|
||||
});
|
||||
|
||||
assert_eq!(policy.schema_version, "cultnet.reconnect_policy.v0");
|
||||
assert_eq!(policy.policy_id, "rudp-default");
|
||||
assert_eq!(policy.max_attempts, Some(8));
|
||||
assert_eq!(compute_reconnect_delay_ms(&policy, 1, 0), 1_000);
|
||||
assert_eq!(compute_reconnect_delay_ms(&policy, 3, 17), 4_017);
|
||||
assert_eq!(compute_reconnect_delay_ms(&policy, 9, 999), 30_250);
|
||||
assert_eq!(compute_reconnect_delay_ms(&policy, 0, 0), 1_000);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rudp_session_handshake_acks_reliable_connect_and_accept_packets() -> Result<()> {
|
||||
let mut client = CultNetRudpSession::new(CultNetRudpSessionOptions {
|
||||
|
|
|
|||
|
|
@ -58,7 +58,11 @@ export { encodeFrame, LengthPrefixedMessageFramer } from "./framing";
|
|||
export { CultNetPeer, type CultNetPeerEvents } from "./peer";
|
||||
export {
|
||||
TcpFramedTransportConnection,
|
||||
computeCultNetReconnectDelayMs,
|
||||
createCultNetReconnectPolicy,
|
||||
createTcpFramedTransportProfile,
|
||||
type CultNetReconnectPolicy,
|
||||
type CultNetReconnectPolicyOptions,
|
||||
type CultNetTransportConnection,
|
||||
type CultNetTransportConnectionEvents,
|
||||
type CultNetTransportFrame,
|
||||
|
|
|
|||
|
|
@ -16,6 +16,48 @@ export interface CultNetTransportStats {
|
|||
framesSent: number;
|
||||
}
|
||||
|
||||
export interface CultNetReconnectPolicy {
|
||||
schemaVersion: "cultnet.reconnect_policy.v0";
|
||||
policyId: string;
|
||||
baseDelayMs: number;
|
||||
maxDelayMs: number;
|
||||
maxJitterMs: number;
|
||||
maxAttempts?: number;
|
||||
}
|
||||
|
||||
export interface CultNetReconnectPolicyOptions {
|
||||
policyId?: string;
|
||||
baseDelayMs?: number;
|
||||
maxDelayMs?: number;
|
||||
maxJitterMs?: number;
|
||||
maxAttempts?: number;
|
||||
}
|
||||
|
||||
export function createCultNetReconnectPolicy(options: CultNetReconnectPolicyOptions = {}): CultNetReconnectPolicy {
|
||||
return {
|
||||
schemaVersion: "cultnet.reconnect_policy.v0",
|
||||
policyId: options.policyId ?? "default",
|
||||
baseDelayMs: options.baseDelayMs ?? 1_000,
|
||||
maxDelayMs: options.maxDelayMs ?? 30_000,
|
||||
maxJitterMs: options.maxJitterMs ?? 250,
|
||||
...(options.maxAttempts !== undefined ? { maxAttempts: options.maxAttempts } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
export function computeCultNetReconnectDelayMs(
|
||||
policy: CultNetReconnectPolicy,
|
||||
attempt: number,
|
||||
jitterMs = 0,
|
||||
): number {
|
||||
const normalizedAttempt = Math.max(1, Math.floor(attempt));
|
||||
const cappedBaseDelay = Math.min(
|
||||
policy.maxDelayMs,
|
||||
policy.baseDelayMs * (2 ** (normalizedAttempt - 1)),
|
||||
);
|
||||
const boundedJitter = Math.max(0, Math.min(policy.maxJitterMs, Math.floor(jitterMs)));
|
||||
return cappedBaseDelay + boundedJitter;
|
||||
}
|
||||
|
||||
export interface CultNetTransportConnectionEvents {
|
||||
frame: (frame: CultNetTransportFrame) => void;
|
||||
close: () => void;
|
||||
|
|
|
|||
|
|
@ -25,6 +25,8 @@ import {
|
|||
TcpFramedTransportConnection,
|
||||
cultNetSchemas,
|
||||
cultNetBuiltinSchemaRegistry,
|
||||
computeCultNetReconnectDelayMs,
|
||||
createCultNetReconnectPolicy,
|
||||
createTcpFramedTransportProfile,
|
||||
createRudpTransportProfile,
|
||||
defineCultNetDocumentBinding,
|
||||
|
|
@ -275,6 +277,18 @@ test("rudp transport profile advertises state and realtime channel semantics", (
|
|||
);
|
||||
});
|
||||
|
||||
test("reconnect policy exposes the shared portable delay contract", () => {
|
||||
const policy = createCultNetReconnectPolicy({ policyId: "rudp-default", maxAttempts: 8 });
|
||||
|
||||
assert.equal(policy.schemaVersion, "cultnet.reconnect_policy.v0");
|
||||
assert.equal(policy.policyId, "rudp-default");
|
||||
assert.equal(policy.maxAttempts, 8);
|
||||
assert.equal(computeCultNetReconnectDelayMs(policy, 1), 1_000);
|
||||
assert.equal(computeCultNetReconnectDelayMs(policy, 3, 17), 4_017);
|
||||
assert.equal(computeCultNetReconnectDelayMs(policy, 9, 999), 30_250);
|
||||
assert.equal(computeCultNetReconnectDelayMs(policy, 0, -5), 1_000);
|
||||
});
|
||||
|
||||
test("rudp session handshake acks reliable connect and accept packets", () => {
|
||||
const client = new CultNetRudpSession({ connectionId: 0x0a0b0c0d, initialSequence: 1, resendDelayMs: 50 });
|
||||
const server = new CultNetRudpSession({ connectionId: 0x0a0b0c0d, initialSequence: 100, resendDelayMs: 50 });
|
||||
|
|
|
|||
|
|
@ -60,6 +60,84 @@ namespace GameCult.Networking
|
|||
public byte[] Payload { get; set; } = Array.Empty<byte>();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Portable reconnect backoff policy document.
|
||||
/// </summary>
|
||||
[MessagePack.MessagePackObject]
|
||||
public sealed class CultNetReconnectPolicy
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets or sets the shared reconnect policy schema version.
|
||||
/// </summary>
|
||||
[MessagePack.Key("schemaVersion")]
|
||||
public string SchemaVersion { get; set; } = "cultnet.reconnect_policy.v0";
|
||||
/// <summary>
|
||||
/// Gets or sets the policy identifier.
|
||||
/// </summary>
|
||||
[MessagePack.Key("policyId")]
|
||||
public string PolicyId { get; set; } = "default";
|
||||
/// <summary>
|
||||
/// Gets or sets the first reconnect delay.
|
||||
/// </summary>
|
||||
[MessagePack.Key("baseDelayMs")]
|
||||
public int BaseDelayMs { get; set; } = 1_000;
|
||||
/// <summary>
|
||||
/// Gets or sets the maximum exponential backoff delay before jitter.
|
||||
/// </summary>
|
||||
[MessagePack.Key("maxDelayMs")]
|
||||
public int MaxDelayMs { get; set; } = 30_000;
|
||||
/// <summary>
|
||||
/// Gets or sets the maximum positive jitter a caller may add.
|
||||
/// </summary>
|
||||
[MessagePack.Key("maxJitterMs")]
|
||||
public int MaxJitterMs { get; set; } = 250;
|
||||
/// <summary>
|
||||
/// Gets or sets the optional maximum reconnect attempts.
|
||||
/// </summary>
|
||||
[MessagePack.Key("maxAttempts")]
|
||||
public int? MaxAttempts { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Helpers for portable reconnect policy documents.
|
||||
/// </summary>
|
||||
public static class CultNetReconnectPolicies
|
||||
{
|
||||
/// <summary>
|
||||
/// Creates a reconnect policy using the shared default values.
|
||||
/// </summary>
|
||||
public static CultNetReconnectPolicy CreateDefault(
|
||||
string policyId = "default",
|
||||
int baseDelayMs = 1_000,
|
||||
int maxDelayMs = 30_000,
|
||||
int maxJitterMs = 250,
|
||||
int? maxAttempts = null)
|
||||
{
|
||||
return new CultNetReconnectPolicy
|
||||
{
|
||||
PolicyId = string.IsNullOrWhiteSpace(policyId) ? "default" : policyId,
|
||||
BaseDelayMs = baseDelayMs,
|
||||
MaxDelayMs = maxDelayMs,
|
||||
MaxJitterMs = maxJitterMs,
|
||||
MaxAttempts = maxAttempts
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Computes the deterministic exponential reconnect delay for an attempt.
|
||||
/// </summary>
|
||||
public static int ComputeDelayMs(CultNetReconnectPolicy policy, int attempt, int jitterMs = 0)
|
||||
{
|
||||
if (policy == null) throw new ArgumentNullException(nameof(policy));
|
||||
var normalizedAttempt = Math.Max(1, attempt);
|
||||
var cappedBaseDelay = Math.Min(
|
||||
policy.MaxDelayMs,
|
||||
(int)Math.Min(int.MaxValue, policy.BaseDelayMs * Math.Pow(2, normalizedAttempt - 1)));
|
||||
var boundedJitter = Math.Max(0, Math.Min(policy.MaxJitterMs, jitterMs));
|
||||
return cappedBaseDelay + boundedJitter;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Options for creating a TCP framed transport profile.
|
||||
/// </summary>
|
||||
|
|
|
|||
|
|
@ -390,6 +390,26 @@ namespace GameCult.Networking.Tests
|
|||
Is.EqualTo(new int?[] { 64, 64, 64 }));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void ReconnectPolicy_ExposesPortableDelayContract()
|
||||
{
|
||||
var policy = CultNetReconnectPolicies.CreateDefault("rudp-default", maxAttempts: 8);
|
||||
|
||||
Assert.That(policy.SchemaVersion, Is.EqualTo("cultnet.reconnect_policy.v0"));
|
||||
Assert.That(policy.PolicyId, Is.EqualTo("rudp-default"));
|
||||
Assert.That(policy.MaxAttempts, Is.EqualTo(8));
|
||||
Assert.That(CultNetReconnectPolicies.ComputeDelayMs(policy, 1), Is.EqualTo(1_000));
|
||||
Assert.That(CultNetReconnectPolicies.ComputeDelayMs(policy, 3, 17), Is.EqualTo(4_017));
|
||||
Assert.That(CultNetReconnectPolicies.ComputeDelayMs(policy, 9, 999), Is.EqualTo(30_250));
|
||||
Assert.That(CultNetReconnectPolicies.ComputeDelayMs(policy, 0, -5), Is.EqualTo(1_000));
|
||||
|
||||
var json = MessagePack.MessagePackSerializer.ConvertToJson(
|
||||
MessagePack.MessagePackSerializer.Serialize(policy, CultNetSchemaMessageSerialization.Options));
|
||||
Assert.That(json, Does.Contain("\"schemaVersion\""));
|
||||
Assert.That(json, Does.Contain("\"policyId\""));
|
||||
Assert.That(json, Does.Contain("\"maxAttempts\""));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void RudpSession_HandshakeAcksReliableConnectAndAcceptPackets()
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in a new issue