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 7d052785..10cc2a58 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 @@ -808,7 +808,6 @@ interface CultNetSchemaMessageTransport : AutoCloseable { val profile: CultNetTransportProfile val stats: CultNetTransportStats - /** Fire-and-forget: an ack or a return is not admission; read the result frame. */ fun sendSchemaMessage(message: CultNetMessage) fun receiveSchemaMessage(timeoutMs: Long = 1_000, pollIntervalMs: Long = 5): CultNetMessage? @@ -3168,16 +3167,13 @@ class CultNetRudpSocketTransportConnection( return connected } - /** Fire-and-forget: an ack or a return is not admission; read the result frame. */ fun send(channelId: String, payload: ByteArray) { session.sendMany(channelId, payload, channelSendOptions(channelId, nowMs()), maxFragmentBytes).forEach { sendPacket(it) } framesSent += 1 } - /** Fire-and-forget: an ack or a return is not admission; read the result frame. */ fun send(channelId: String, payload: String) = send(channelId, payload.toByteArray(StandardCharsets.UTF_8)) - /** Fire-and-forget: an ack or a return is not admission; read the result frame. */ fun sendSchema(payload: ByteArray) = send("schema", payload) fun sendSchema(payload: String) = sendSchema(payload.toByteArray(StandardCharsets.UTF_8)) diff --git a/packages/cultmesh-py/src/cultmesh_py/server.py b/packages/cultmesh-py/src/cultmesh_py/server.py index ab5267a7..69583c1c 100644 --- a/packages/cultmesh-py/src/cultmesh_py/server.py +++ b/packages/cultmesh-py/src/cultmesh_py/server.py @@ -59,16 +59,6 @@ from cultnet_py.cultmesh_contracts import ( ) _LOGGER = logging.getLogger(__name__) - -# How many consecutive refused frames of one kind a RUDP session may send before it is ended: undecodable -# frames (reset by any frame that decodes) and deletes the write authority refuses (reset by any decoded -# frame that is not a refused delete). Fewer are answered with a typed refusal and the session is kept; the -# frame that reaches this number is answered too, then the session ends. The same number is the cap in every -# runtime. -MAX_CONSECUTIVE_REFUSED_FRAMES = 8 - -# How long a session whose capped frame was answered waits for the peer to acknowledge the answer before it ends anyway. -_RUDP_CAPPED_ANSWER_GRACE_MS = 2000 _EPHEMERAL_PAIR_ATTEMPTS = 64 @@ -100,10 +90,6 @@ class _DatabaseSubscription: class _RudpPeerConnection: session: CultNetRudpSession subscriptions: dict[str, _DatabaseSubscription] = field(default_factory=dict) - undecodable_frames_in_a_row: int = 0 - refused_deletes_in_a_row: int = 0 - # Set once the peer's capped frame has been answered: the session ends when the answers are acknowledged. - ending_deadline_ms: int | None = None @dataclass @@ -360,7 +346,6 @@ class CultMeshLocalServer: if packet.connection_id != self.rudp_connection_id: continue - self._end_answered_peers(rudp_socket, peers) now_ms = _now_ms() peer = peers.get(remote_addr) if packet.packet_type == CultNetRudpPacketType.CONNECT: @@ -409,31 +394,18 @@ class CultMeshLocalServer: for delivered in result.delivered: if delivered.channel_id != "schema": continue - if peer.ending_deadline_ms is not None: - # The capped frame was answered; nothing after it is served while the answer is delivered. - break try: message = msgpack.unpackb(delivered.payload, raw=False) except (msgpack.ExtraData, msgpack.FormatError, msgpack.StackError, ValueError): - message = None - if isinstance(message, dict): - peer.undecodable_frames_in_a_row = 0 - responses = self._handle_connection_message( - message, - peer.subscriptions, - over_rudp=True, - peer=CultMeshServerPeer(f"{remote_addr[0]}:{remote_addr[1]}", "rudp"), - ) - refused_delete = message.get("schemaVersion") == "cultnet.document_delete.v0" and any( - response.get("code") == "write_not_authorized" for response in responses - ) - peer.refused_deletes_in_a_row = peer.refused_deletes_in_a_row + 1 if refused_delete else 0 - refused, run = refused_delete, peer.refused_deletes_in_a_row - else: - # A frame that is no CultNet message is answered, not dropped, and the session kept. - peer.undecodable_frames_in_a_row += 1 - refused, run = True, peer.undecodable_frames_in_a_row - responses = [self._message_invalid({}, "message", "CultMesh RUDP server could not read the message.")] + continue + if not isinstance(message, dict): + continue + responses = self._handle_connection_message( + message, + peer.subscriptions, + over_rudp=True, + peer=CultMeshServerPeer(f"{remote_addr[0]}:{remote_addr[1]}", "rudp"), + ) for response in responses: if unsendable is None: unsendable = self._send_rudp_schema_frame( @@ -443,11 +415,6 @@ class CultMeshLocalServer: msgpack.packb(response, use_bin_type=True), message_id=str(response.get("messageId") or ""), ) - if refused and run >= MAX_CONSECUTIVE_REFUSED_FRAMES: - # The capped frame is answered like any other, reliably. The session ends once the peer - # has acknowledged every answer (or the grace runs out): see _end_answered_peers. - peer.ending_deadline_ms = now_ms + _RUDP_CAPPED_ANSWER_GRACE_MS - break except Exception as error: # noqa: BLE001 - one peer's message never ends the thread # Whatever a peer's message makes the handling raise is that peer's failure. # The session recorded the frame's sequence, so it cannot be kept: a retransmit @@ -478,26 +445,11 @@ class CultMeshLocalServer: if unsendable is not None: self._end_unsendable_peer(rudp_socket, peers, remote_addr, unsendable) - def _end_answered_peers( - self, - rudp_socket: socket.socket, - peers: dict[tuple[str, int], _RudpPeerConnection], - ) -> None: - """Ends each session whose capped frame was answered once the peer has acknowledged every answer (a lost - answer is resent first), or when the grace has run out.""" - now_ms = _now_ms() - for remote_addr, peer in list(peers.items()): - if peer.ending_deadline_ms is None: - continue - if peer.session.outstanding_reliable_packet_count == 0 or now_ms >= peer.ending_deadline_ms: - self._end_refused_peer(rudp_socket, peers, remote_addr) - def _poll_rudp_resends( self, rudp_socket: socket.socket, peers: dict[tuple[str, int], _RudpPeerConnection], ) -> None: - self._end_answered_peers(rudp_socket, peers) now_ms = _now_ms() for remote_addr, peer in list(peers.items()): for packet in peer.session.due_resends(now_ms): diff --git a/packages/cultmesh-py/tests/test_client_write_completion.py b/packages/cultmesh-py/tests/test_client_write_completion.py deleted file mode 100644 index 0f9d0c70..00000000 --- a/packages/cultmesh-py/tests/test_client_write_completion.py +++ /dev/null @@ -1,52 +0,0 @@ -"""Heimdall's invariant: no client treats a reliable-transport ack as write admission. A put completes only on -its result frame (write_not_authorized, or the success reply). - -The Python write path is a subscription over RUDP: `send` queues the put, `read_next` returns the result frame -and raises the peer's refusal. The server here refuses the put; its reliable ack of the put arrives (the test -waits for it) and the client still reports the refusal, never success. The transport's `receive` yields data -frames only, so the order of the ack against the refusal cannot change what `read_next` reports. -""" - -from __future__ import annotations - -import unittest - -from cultcache_py import define_database_entry_type -from cultmesh_py import CultMesh, create_node -from cultmesh_py.server import CultMeshLocalServer -from cultnet_py import CultNetDatabaseSubscription, CultNetPeerError, create_rudp_schema_transport - -from test_remote_write_authority import KEY_A, NOTE, SCHEMA_ID, put_of - - -class ClientWriteCompletionTests(unittest.TestCase): - def test_a_put_the_server_refused_is_reported_refused_after_its_ack_arrived(self) -> None: - node = create_node(runtime_id="completion-server") - node.database.register_document(NOTE) - server = CultMeshLocalServer(node=node, port=0, write_authority=None) - server.start() - try: - subscription = CultNetDatabaseSubscription( - host="127.0.0.1", - port=server.port, - timeout_seconds=4.0, - subscription_id="completion", - schema_ids=[SCHEMA_ID], - include_snapshot=False, - create_transport=lambda: create_rudp_schema_transport( - host="127.0.0.1", port=server.port, connection_id=server.rudp_connection_id, timeout_seconds=4.0 - ), - ) - with subscription as sub: - sub.send(put_of(KEY_A, "peer")) - sub._transport.flush_reliable(2.0) # the put's reliable ack has arrived - with self.assertRaises(CultNetPeerError) as raised: - sub.read_next() - finally: - server.stop() - self.assertEqual(raised.exception.response["code"], "write_not_authorized") - self.assertIsNone(server.node.cache.get(NOTE, KEY_A)) - - -if __name__ == "__main__": - unittest.main() diff --git a/packages/cultmesh-py/tests/test_cultmesh.py b/packages/cultmesh-py/tests/test_cultmesh.py index d96c4ab2..b49b6ed9 100644 --- a/packages/cultmesh-py/tests/test_cultmesh.py +++ b/packages/cultmesh-py/tests/test_cultmesh.py @@ -24,9 +24,7 @@ from cultcache_py import ( define_document_type, ) from cultcache_py.interop import INTEROP_SCHEMA_VERSION, interop_note_document -from cultmesh_py.server import MAX_CONSECUTIVE_REFUSED_FRAMES from cultmesh_py.verify import verify -from cultmesh_py.write_authority import CultMeshRemoteWriteDecision from cultnet_py import ( compute_simulation_claim_hash, CultNetClientAuthorityScope, @@ -4891,140 +4889,6 @@ class CultMeshTests(unittest.TestCase): self.assertEqual(after["documents"], []) self.assertTrue(alive) - def _rudp_open(self, server: Any) -> Any: - return create_rudp_schema_transport( - host="127.0.0.1", - port=server.port, - connection_id=server.rudp_connection_id, - timeout_seconds=4.0, - max_fragment_bytes=1200, - ) - - def _rudp_answer(self, transport: Any, payload: bytes) -> dict[str, Any]: - transport.send("schema", payload) - return msgpack.unpackb(transport.receive(timeout_seconds=4.0).payload, raw=False) - - def _assert_rudp_session_ended(self, transport: Any, server: Any) -> None: - transport.send("schema", msgpack.packb(snapshot_request(message_id="after-cap").to_wire(), use_bin_type=True)) - with self.assertRaises(Exception): - transport.receive(timeout_seconds=1.0) - self.assertTrue(server._rudp_thread.is_alive()) - - def test_cultmesh_local_server_answers_unreadable_frames_and_ends_the_session_at_eight_in_a_row(self) -> None: - # An undecodable frame (not a msgpack map) is answered message_invalid and the session kept; - # eight in a row end it (the eighth is answered too); any decoded frame resets the run. - self.assertEqual(MAX_CONSECUTIVE_REFUSED_FRAMES, 8) - node = CultMesh.create_node(runtime_id="mesh-rudp-unreadable") - server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites()) - garbage = b"\xc1" - snapshot = msgpack.packb(snapshot_request(message_id="ok").to_wire(), use_bin_type=True) - transport = self._rudp_open(server) - try: - for _ in range(2): - for _ in range(MAX_CONSECUTIVE_REFUSED_FRAMES - 1): - reply = self._rudp_answer(transport, garbage) - self.assertEqual(reply["code"], "message_invalid") - self.assertEqual(reply["details"]["field"], "message") - self.assertEqual(self._rudp_answer(transport, snapshot)["messageId"], "ok") - for _ in range(MAX_CONSECUTIVE_REFUSED_FRAMES - 1): - self._rudp_answer(transport, garbage) - transport.send("schema", garbage) - eighth = msgpack.unpackb(transport.receive(timeout_seconds=4.0).payload, raw=False) - self.assertEqual(eighth["code"], "message_invalid") - self._assert_rudp_session_ended(transport, server) - finally: - transport.close() - server.stop() - - def test_cultmesh_local_server_resends_a_lost_capped_answer_before_it_ends_the_session(self) -> None: - # The eighth refusal is answered reliably: the datagram carrying it is lost on the wire, and the - # peer still gets the answer (by resend) before the goodbye. An unreliable final answer, or a - # goodbye sent at once, would leave the peer without it. - node = CultMesh.create_node(runtime_id="mesh-rudp-lost-answer") - server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites()) - send = server._send_rudp_packet - dropped: list[Any] = [] - armed = threading.Event() - - def lossy(rudp_socket: Any, remote_addr: Any, packet: Any) -> Any: - if armed.is_set() and not dropped and packet.packet_type == CultNetRudpPacketType.DATA and b"message_invalid" in (packet.payload or b""): - dropped.append(packet) - return None - return send(rudp_socket, remote_addr, packet) - - server._send_rudp_packet = lossy - transport = self._rudp_open(server) - try: - for _ in range(MAX_CONSECUTIVE_REFUSED_FRAMES - 1): - self._rudp_answer(transport, b"\xc1") - armed.set() - reply = self._rudp_answer(transport, b"\xc1") - self.assertEqual(len(dropped), 1, "fixture: the first transmission of the last answer was lost") - self.assertEqual(reply["code"], "message_invalid") - self._assert_rudp_session_ended(transport, server) - finally: - transport.close() - server.stop() - - def test_cultmesh_local_server_serves_nothing_after_the_capped_frame(self) -> None: - node = CultMesh.create_node(runtime_id="mesh-rudp-after-cap") - server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites()) - transport = self._rudp_open(server) - try: - for _ in range(MAX_CONSECUTIVE_REFUSED_FRAMES - 1): - self._rudp_answer(transport, b"\xc1") - transport.send("schema", b"\xc1") - transport.send("schema", msgpack.packb(snapshot_request(message_id="behind-the-cap").to_wire(), use_bin_type=True)) - first = msgpack.unpackb(transport.receive(timeout_seconds=4.0).payload, raw=False) - self.assertEqual(first["code"], "message_invalid") - with self.assertRaises(Exception): - transport.receive(timeout_seconds=1.0) - finally: - transport.close() - server.stop() - - def test_cultmesh_local_server_ends_the_session_at_eight_refused_deletes_in_a_row(self) -> None: - class GrantOnlyKeyB: - def decide(self, request: Any) -> Any: - return CultMeshRemoteWriteDecision.grant() if request.record_key == "note:b" else CultMeshRemoteWriteDecision.refuse("no") - - document = self._rudp_bound_note() - node = CultMesh.create_node(runtime_id="mesh-rudp-deletes") - node.database.register_document(document) - server = CultMesh.serve_node(node, write_authority=GrantOnlyKeyB()) - - def delete(key: str) -> bytes: - return msgpack.packb({ - "schemaVersion": "cultnet.document_delete.v0", - "messageId": f"del-{key}", - "schemaId": document.schema_id, - "recordKey": key, - }, use_bin_type=True) - - snapshot = msgpack.packb(snapshot_request(message_id="ok").to_wire(), use_bin_type=True) - transport = self._rudp_open(server) - try: - def refused_run(count: int) -> None: - for _ in range(count): - reply = self._rudp_answer(transport, delete("note:a")) - self.assertEqual(reply["code"], "write_not_authorized") - self.assertEqual(reply["details"]["value"], "note:a") - - refused_run(MAX_CONSECUTIVE_REFUSED_FRAMES - 1) - # A granted delete resets the run (nothing is stored under it, so the store answers that). - self.assertEqual(self._rudp_answer(transport, delete("note:b"))["code"], "unregistered_document_delete") - refused_run(MAX_CONSECUTIVE_REFUSED_FRAMES - 1) - # So does any decoded non-delete frame. - self.assertEqual(self._rudp_answer(transport, snapshot)["messageId"], "ok") - refused_run(MAX_CONSECUTIVE_REFUSED_FRAMES - 1) - transport.send("schema", delete("note:a")) - eighth = msgpack.unpackb(transport.receive(timeout_seconds=4.0).payload, raw=False) - self.assertEqual(eighth["code"], "write_not_authorized") - self._assert_rudp_session_ended(transport, server) - finally: - transport.close() - server.stop() - def test_cultmesh_local_server_refuses_a_put_its_store_cannot_take_and_serves_on(self) -> None: # Well-formed puts the store refuses: a blank key, a payload that is not MessagePack, a # payload its schema cannot decode, a global document under another key. Each is refused diff --git a/packages/cultmesh-rs/src/lib.rs b/packages/cultmesh-rs/src/lib.rs index 33cb4858..d7c2c8fd 100644 --- a/packages/cultmesh-rs/src/lib.rs +++ b/packages/cultmesh-rs/src/lib.rs @@ -82,10 +82,7 @@ pub struct CultMeshRudpDocumentPublishOptions { pub target: SocketAddr, pub runtime_id: String, pub connect_timeout: Duration, - /// How long the acknowledgement of every put may take. pub flush_timeout: Duration, - /// How long after the acknowledgement a refusal may still arrive; a success frame ends the wait early. - pub result_window: Duration, pub poll_interval: Duration, pub resend_delay_ms: u64, pub source_agent_id: Option, @@ -152,7 +149,6 @@ impl Default for CultMeshRudpDocumentPublishOptions { runtime_id: "cultmesh-rudp-document-publisher".to_string(), connect_timeout: Duration::from_secs(3), flush_timeout: Duration::from_millis(300), - result_window: Duration::from_millis(500), poll_interval: Duration::from_millis(10), resend_delay_ms: 50, source_agent_id: None, @@ -886,30 +882,77 @@ fn unix_millis() -> u128 { /// Publish one raw CultNet schema message to a RUDP catalog. /// -/// A put completes on the catalog's result frame, never on the transport -/// acknowledgement: an ack proves delivery, not admission. A refusal (a -/// `cultnet.error.v0` frame) returns an error in either order against the ack, -/// naming the refusal when it is one of the fixed texts a catalog sends. A -/// catalog that admits a put sends no success frame, so after the ack the call -/// waits `result_window` for a refusal and reports the put as admitted when -/// none comes (interim until the catalog sends a success frame; a success -/// frame ends the wait early). Each call owns a fresh connection epoch. +/// Each call owns a fresh connection epoch and returns only after the catalog +/// acknowledges reliable application admission. Against a Rust CultMesh +/// catalog, which never acknowledges a message it refuses, a refusal is +/// returned as an error, naming the refusal when it is one of the fixed texts +/// such a catalog sends. A catalog that acknowledges a message on receipt and +/// refuses it afterwards (the TypeScript document server does) is reported as +/// having admitted it: the acknowledgement arrives first. pub fn publish_cultnet_message_to_rudp_catalog( message: &cultnet_rs::CultNetMessage, options: CultMeshRudpDocumentPublishOptions, ) -> Result<()> { - publish_cultnet_messages_to_rudp_catalog(std::slice::from_ref(message), options) + if options.runtime_id.trim().is_empty() { + anyhow::bail!("runtime_id must be non-empty"); + } + let mut client = connect_rudp_client( + options.target, + &options.runtime_id, + options.connect_timeout, + options.poll_interval, + options.resend_delay_ms, + "cultmesh-rudp-document-publisher", + )?; + + let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?; + let receipt = client.send_reliable("schema", payload)?; + let flush_deadline = Instant::now() + options.flush_timeout; + while Instant::now() < flush_deadline { + if let Some(frame) = client.receive_once()? { + if frame.channel_id == "schema" { + if let Ok(cultnet_rs::CultNetMessage::Error { error, .. }) = + decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + ) + { + let _ = client.disconnect(b"document-refused".to_vec()); + anyhow::bail!( + "CultMesh RUDP catalog {} refused the put{}", + options.target, + recognised_refusal(&error) + ); + } + } + } + match client.reliable_send_status(&receipt) { + cultnet_rs::CultNetRudpReliableSendStatus::Acknowledged => { + let _ = client.disconnect(b"document-acknowledged".to_vec()); + return Ok(()); + } + cultnet_rs::CultNetRudpReliableSendStatus::Invalidated => { + anyhow::bail!("CultMesh RUDP document receipt was invalidated") + } + cultnet_rs::CultNetRudpReliableSendStatus::Pending => {} + } + client.poll_resends()?; + thread::sleep(options.poll_interval); + } + let _ = client.disconnect(b"document-unacknowledged".to_vec()); + anyhow::bail!( + "timed out waiting for reliable acknowledgement from CultMesh RUDP catalog {}", + options.target + ) } -/// Publish a batch of raw CultNet schema messages to a RUDP catalog on one -/// connection. +/// Publish a batch of raw CultNet schema messages to a RUDP catalog. /// -/// Completes under the same rule as [`publish_cultnet_message_to_rudp_catalog`]: -/// every message acknowledged, then no refusal frame within `result_window` -/// (or one success frame per message). A refusal of any message in the batch -/// fails the whole call; the batch cannot hide it. The catalog's answers do not -/// name the message they answer, so a failed batch does not say which message -/// was refused or whether the others were admitted. +/// Unlike [`publish_cultnet_message_to_rudp_catalog`], which polls for +/// per-message application-level acknowledgement, this sends every message on +/// one connection and waits only for a transport-level `flush_reliable`. It +/// trades per-receipt confirmation for batch throughput; callers that need a +/// stronger delivery guarantee should use the singular publish path instead. pub fn publish_cultnet_messages_to_rudp_catalog( messages: &[cultnet_rs::CultNetMessage], options: CultMeshRudpDocumentPublishOptions, @@ -926,62 +969,11 @@ pub fn publish_cultnet_messages_to_rudp_catalog( "cultmesh-rudp-document-publisher", )?; - let mut receipts = Vec::with_capacity(messages.len()); for message in messages { let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?; - receipts.push(client.send_reliable("schema", payload)?); - } - let ack_deadline = Instant::now() + options.flush_timeout; - let mut acknowledged_at: Option = None; - let mut successes = 0; - loop { - while let Some(frame) = client.receive_once()? { - if frame.channel_id != "schema" { - continue; - } - match decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0) { - Ok(cultnet_rs::CultNetMessage::Error { error, .. }) => { - let _ = client.disconnect(b"document-refused".to_vec()); - anyhow::bail!( - "CultMesh RUDP catalog {} refused the put{}", - options.target, - recognised_refusal(&error) - ); - } - Ok(_) => successes += 1, - Err(_) => {} - } - } - if acknowledged_at.is_none() { - let mut all_acknowledged = true; - for receipt in &receipts { - match client.reliable_send_status(receipt) { - cultnet_rs::CultNetRudpReliableSendStatus::Acknowledged => {} - cultnet_rs::CultNetRudpReliableSendStatus::Invalidated => { - anyhow::bail!("CultMesh RUDP document receipt was invalidated") - } - cultnet_rs::CultNetRudpReliableSendStatus::Pending => all_acknowledged = false, - } - } - if all_acknowledged { - acknowledged_at = Some(Instant::now()); - } else if Instant::now() >= ack_deadline { - let _ = client.disconnect(b"document-unacknowledged".to_vec()); - anyhow::bail!( - "timed out waiting for reliable acknowledgement from CultMesh RUDP catalog {}", - options.target - ) - } - } - if let Some(acknowledged_at) = acknowledged_at { - if successes >= messages.len() || acknowledged_at.elapsed() >= options.result_window { - let _ = client.disconnect(b"document-completed".to_vec()); - return Ok(()); - } - } - client.poll_resends()?; - thread::sleep(options.poll_interval); + client.send("schema", payload)?; } + client.flush_reliable(options.flush_timeout) } fn connect_rudp_client( @@ -1459,141 +1451,6 @@ mod tests { Ok((target, stop, catalog)) } - /// A catalog that acknowledges every reliable message it receives and refuses it with a typed - /// write_not_authorized, the acknowledgement either before the refusal or after it (the second packet - /// follows the first by 100 ms). The refusal carries no acknowledgement of its own. - /// What the stub catalog sends for a put, and in which order against the transport ack. - #[derive(Clone, Copy, PartialEq, Eq)] - enum CatalogReply { - RefusalThenAck, - AckThenRefusal, - AckThenSuccess, - } - - fn acknowledging_refusing_catalog( - reply: CatalogReply, - ) -> Result<( - SocketAddr, - std::sync::Arc, - thread::JoinHandle>, - )> { - use cultnet_rs::{ - CultNetRudpPacketType, CultNetRudpSendOptions, CultNetRudpSession, - CultNetRudpSessionOptions, decode_rudp_packet, encode_rudp_packet, - }; - let socket = UdpSocket::bind("127.0.0.1:0")?; - socket.set_read_timeout(Some(Duration::from_millis(5)))?; - let target = socket.local_addr()?; - let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); - let catalog_stop = stop.clone(); - let catalog = thread::spawn(move || -> Result<()> { - let mut session: Option = None; - let mut wire = vec![0_u8; 65_535]; - while !catalog_stop.load(std::sync::atomic::Ordering::SeqCst) { - let Ok((bytes, peer)) = socket.recv_from(&mut wire) else { - continue; - }; - let Ok(packet) = decode_rudp_packet(&wire[..bytes]) else { - continue; - }; - if packet.packet_type == CultNetRudpPacketType::Connect { - let mut accepted = CultNetRudpSession::new(CultNetRudpSessionOptions { - connection_id: packet.connection_id, - initial_sequence: None, - resend_delay_ms: 10_000, - max_pending_reliable_packets: Some(64), - }); - let accept = accepted.accept_connect(&packet, 0, Vec::new())?; - socket.send_to(&encode_rudp_packet(&accept)?, peer)?; - session = Some(accepted); - continue; - } - let Some(session) = session.as_mut() else { - continue; - }; - let Ok(received) = session.receive(&packet, 1) else { - continue; - }; - if !packet.reliable || received.delivered.is_empty() { - continue; - } - let ack = session.create_ack_for_received(packet.sequence); - let refusal = encode_cultnet_message_to_vec( - &if reply == CatalogReply::AckThenSuccess { - cultnet_rs::CultNetMessage::SnapshotResponseRaw { - message_id: "completion".into(), - documents: Vec::new(), - } - } else { - cultnet_rs::CultNetMessage::Error { - error: "write_not_authorized: the catalog refused the document".into(), - code: Some(cultnet_rs::CultNetErrorCode::WriteNotAuthorized), - details: None, - } - }, - CultNetWireContract::CultNetSchemaV0, - )?; - let options = CultNetRudpSendOptions { - reliable: false, - ordered: false, - sequenced: false, - now_ms: 2, - reliable_expire_after_ms: None, - }; - let mut refusals = session.send_many("schema", refusal, options, Some(1200))?; - for refusal in &mut refusals { - refusal.ack = 0; - refusal.ack_mask = 0; - } - let send_refusals = |socket: &UdpSocket| -> Result<()> { - for refusal in &refusals { - socket.send_to(&encode_rudp_packet(refusal)?, peer)?; - } - Ok(()) - }; - if reply == CatalogReply::RefusalThenAck { - send_refusals(&socket)?; - thread::sleep(Duration::from_millis(100)); - socket.send_to(&encode_rudp_packet(&ack)?, peer)?; - } else { - socket.send_to(&encode_rudp_packet(&ack)?, peer)?; - thread::sleep(Duration::from_millis(100)); - send_refusals(&socket)?; - } - } - Ok(()) - }); - Ok((target, stop, catalog)) - } - - fn publish_options(target: SocketAddr) -> CultMeshRudpDocumentPublishOptions { - CultMeshRudpDocumentPublishOptions { - target, - connect_timeout: Duration::from_secs(2), - flush_timeout: Duration::from_secs(3), - poll_interval: Duration::from_millis(2), - resend_delay_ms: 5, - ..CultMeshRudpDocumentPublishOptions::default() - } - } - - fn refused_note() -> cultnet_rs::CultNetMessage { - cultnet_rs::CultNetMessage::DocumentPutRaw { - message_id: "completion".into(), - document: cultnet_rs::CultNetRawDocumentRecord { - schema_id: "cultmesh.test.raw.v0".into(), - record_key: "tests:completion:a".into(), - stored_at: "2026-11-08T00:00:00Z".into(), - payload_encoding: cultnet_rs::CultNetRawPayloadEncoding::Messagepack, - payload: vec![0xc0], - source_runtime_id: None, - source_agent_id: None, - source_role: None, - tags: None, - }, - } - } - /// A catalog's refusal text is quoted only when it is one of the fixed /// refusals a CultMesh catalog sends. Any other text, which could carry /// anything, stays out of the publisher's error. @@ -2044,63 +1901,4 @@ mod tests { assert_eq!(lease.bytes(), &[1, 2, 3]); Ok(()) } - - // Heimdall's invariant: no client treats a reliable-transport ack as write admission. The catalog here - // acknowledges the put and refuses it, the refusal arriving before or after the ack. - #[test] - fn a_publish_whose_refusal_arrives_before_the_ack_is_refused() -> Result<()> { - let (target, stop, catalog) = acknowledging_refusing_catalog(CatalogReply::RefusalThenAck)?; - let result = publish_cultnet_message_to_rudp_catalog(&refused_note(), publish_options(target)); - stop.store(true, std::sync::atomic::Ordering::SeqCst); - catalog.join().expect("catalog thread should not panic")?; - let error = result.expect_err("a refused put is never reported as admitted"); - assert!(error.to_string().contains("refused the put"), "{error:#}"); - Ok(()) - } - - - #[test] - fn a_publish_whose_ack_arrives_before_the_refusal_is_refused() -> Result<()> { - let (target, stop, catalog) = acknowledging_refusing_catalog(CatalogReply::AckThenRefusal)?; - let result = publish_cultnet_message_to_rudp_catalog(&refused_note(), publish_options(target)); - stop.store(true, std::sync::atomic::Ordering::SeqCst); - catalog.join().expect("catalog thread should not panic")?; - let error = result.expect_err("a refused put is never reported as admitted"); - assert!(error.to_string().contains("refused the put"), "{error:#}"); - Ok(()) - } - - #[test] - fn a_batch_publish_is_not_complete_on_the_ack_of_refused_puts() -> Result<()> { - for reply in [CatalogReply::AckThenRefusal, CatalogReply::RefusalThenAck] { - let (target, stop, catalog) = acknowledging_refusing_catalog(reply)?; - let result = publish_cultnet_messages_to_rudp_catalog(&[refused_note()], publish_options(target)); - stop.store(true, std::sync::atomic::Ordering::SeqCst); - catalog.join().expect("catalog thread should not panic")?; - assert!(result.is_err(), "a refused put was reported as published"); - } - Ok(()) - } - - #[test] - fn a_success_frame_completes_the_publish_without_waiting_out_the_result_window() -> Result<()> { - for batch in [false, true] { - let (target, stop, catalog) = acknowledging_refusing_catalog(CatalogReply::AckThenSuccess)?; - let options = CultMeshRudpDocumentPublishOptions { - result_window: Duration::from_secs(30), - ..publish_options(target) - }; - let started = Instant::now(); - let result = if batch { - publish_cultnet_messages_to_rudp_catalog(&[refused_note()], options) - } else { - publish_cultnet_message_to_rudp_catalog(&refused_note(), options) - }; - stop.store(true, std::sync::atomic::Ordering::SeqCst); - catalog.join().expect("catalog thread should not panic")?; - result?; - assert!(started.elapsed() < Duration::from_secs(10), "batch={batch}: waited out the window"); - } - Ok(()) - } } diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index 647589e0..817e9b5c 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -19,16 +19,6 @@ const MAX_FRAGMENTS_PER_MESSAGE: usize = u16::MAX as usize; /// can carry no shorter id, so no response to it can be smaller. const SHORTEST_MESSAGE_ID: &str = "0"; -/// How many consecutive refused frames of one kind a session may send before it is ended: undecodable -/// frames (reset by any frame that decodes) and refused deletes (reset by any decoded non-delete). -/// Fewer are answered with a typed refusal and the session is kept; the cap that reaches this number -/// is answered too, then the session ends. The same number is the cap in every runtime. -pub const MAX_CONSECUTIVE_REFUSED_FRAMES: u32 = 8; - -/// How long a session whose capped frame was answered waits for the peer to acknowledge the answer -/// before it ends anyway. -const CAPPED_ANSWER_GRACE_MILLIS: u64 = 2_000; - /// The transport identity of one remote CultNet RUDP session. /// /// Connection identifiers are scoped to a remote socket address. Two clients @@ -206,10 +196,7 @@ pub struct CultMeshRudpMaintenance { #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum CultMeshRudpApplicationOperation { DocumentPutRaw, - DocumentDelete, SnapshotRequest, - /// A schema frame that did not decode as any CultNet message; there is no message id to name. - Unreadable, } /// Why the server refused an application message. @@ -254,18 +241,12 @@ pub enum CultMeshRudpRejectionReason { /// (`send_error_code`). The session has ended with a goodbye carrying the /// same name, and no refusal: the server knows the peer cannot be reached. ResponseSendFailed(&'static str), - /// The server has no delete path, so every delete is refused. Nothing reached the sink. - DeleteNotSupported, - /// A schema frame did not decode as a CultNet message. Carries the decoder's error text. - MessageUnreadable(String), } impl std::fmt::Display for CultMeshRudpRejectionReason { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::SinkRefused(error) | Self::SnapshotSourceFailed(error) => f.write_str(error), - Self::DeleteNotSupported => f.write_str("the server does not take deletes"), - Self::MessageUnreadable(error) => write!(f, "message could not be read: {error}"), Self::DocumentUnservable { response_bytes, max_snapshot_response_bytes, @@ -306,7 +287,7 @@ impl std::fmt::Display for CultMeshRudpRejectionReason { /// The texts a refused peer is sent, one per rejection variant. They are fixed: /// a sink's or source's error text can quote what a peer sent, so it stays with /// the caller that receives the rejection and never reaches the wire. -pub(crate) const REFUSAL_TEXTS: [&str; 11] = [ +pub(crate) const REFUSAL_TEXTS: [&str; 9] = [ "the catalog refused the document", "the document can never be served", "the snapshot source failed", @@ -316,8 +297,6 @@ pub(crate) const REFUSAL_TEXTS: [&str; 11] = [ "the retained payload budget is full", "the snapshot response could not be queued", "the snapshot response could not be sent", - "the server does not take deletes", - "the message could not be read", ]; impl CultMeshRudpRejectionReason { @@ -333,8 +312,6 @@ impl CultMeshRudpRejectionReason { Self::PayloadBudgetFull => 6, Self::ResponseQueueFailed(_) => 7, Self::ResponseSendFailed(_) => 8, - Self::DeleteNotSupported => 9, - Self::MessageUnreadable(_) => 10, }] } } @@ -354,46 +331,25 @@ pub(crate) fn write_refused_message(record_key: &str, reason: &str) -> CultNetMe } } -/// The typed `message_invalid` refusal naming the request `field` at fault, as C# -/// `CultNetErrorMessage.ForMessageInvalid` builds it. -pub(crate) fn message_invalid_message(field: &str, reason: &str) -> CultNetMessage { - CultNetMessage::Error { - error: format!("message_invalid: {reason}"), - code: Some(CultNetErrorCode::MessageInvalid), - details: Some(CultNetErrorDetails { - field: Some(field.into()), - value: None, - as_of: None, - current: None, - }), - } -} - #[derive(Clone, Debug, PartialEq, Eq)] pub struct CultMeshRudpApplicationRejection { pub session: CultMeshRudpSessionKey, pub operation: CultMeshRudpApplicationOperation, pub message_id: String, - /// The record a rejected put or delete concerned; `None` for a snapshot request or an unreadable frame. - /// A sink or delete refusal names it to the peer. + /// The record a rejected put concerned; `None` for a snapshot request. A sink refusal names it to the peer. pub record_key: Option, pub reason: CultMeshRudpRejectionReason, } impl CultMeshRudpApplicationRejection { - /// What the refused peer is sent: a refused put or delete is a typed `write_not_authorized` naming - /// the record, an unreadable frame a typed `message_invalid` naming `message`; every other rejection - /// is the variant's fixed text, untyped. None carries the sink's, source's or decoder's own text. + /// What the refused peer is sent: a sink's refusal of a put is a typed `write_not_authorized` naming + /// the record; every other rejection is the variant's fixed text, untyped. Neither carries the + /// sink's or source's own text. fn refusal_message(&self) -> CultNetMessage { let text = self.reason.refusal_text(); match (&self.reason, &self.record_key) { - ( - CultMeshRudpRejectionReason::SinkRefused(_) - | CultMeshRudpRejectionReason::DeleteNotSupported, - Some(record_key), - ) => write_refused_message(record_key, text), - (CultMeshRudpRejectionReason::MessageUnreadable(_), _) => { - message_invalid_message("message", text) + (CultMeshRudpRejectionReason::SinkRefused(_), Some(record_key)) => { + write_refused_message(record_key, text) } _ => CultNetMessage::Error { error: text.into(), @@ -426,11 +382,6 @@ struct SessionEntry { created_at_monotonic_millis: u64, last_activity_monotonic_millis: u64, admitted_payload_bytes: usize, - undecodable_frames_in_a_row: u32, - refused_deletes_in_a_row: u32, - /// Set once the peer's capped frame has been answered: the session ends when the answer is acknowledged, - /// or at this time. - ending_deadline_monotonic_millis: Option, } /// A synchronous, multi-session CultNet RUDP raw-document server. @@ -605,18 +556,7 @@ where }; let mut unsendable = match result.reply { - Some(mut reply) => { - // A reply (a Pong) in the grace after the capped frame acknowledges nothing either. - if self - .sessions - .get(&key) - .is_some_and(|entry| entry.ending_deadline_monotonic_millis.is_some()) - { - reply.ack = 0; - reply.ack_mask = 0; - } - self.send_packet(key.remote_addr, &reply)? - } + Some(reply) => self.send_packet(key.remote_addr, &reply)?, None => None, }; @@ -625,125 +565,24 @@ where return Ok(CultMeshRudpPollOutcome::Handled); } - let mut kept_rejection = None; for frame in result.delivered { if frame.channel_id != "schema" { continue; } - // The capped frame was answered: nothing after it is served while the answer is delivered. - if self - .sessions - .get(&key) - .is_some_and(|entry| entry.ending_deadline_monotonic_millis.is_some()) - { - break; - } - let decoded = decode_cultnet_message_from_slice( + let message = match decode_cultnet_message_from_slice( &frame.payload, CultNetWireContract::CultNetSchemaV0, - ); - // Runs of refusals this frame extends or breaks. A session is kept through a short run and - // ended by `MAX_CONSECUTIVE_REFUSED_FRAMES` in a row. - let (is_unreadable, is_delete) = ( - decoded.is_err(), - matches!(decoded, Ok(CultNetMessage::DocumentDelete { .. })), - ); - let run = self.sessions.get_mut(&key).map(|entry| { - entry.undecodable_frames_in_a_row = if is_unreadable { - entry.undecodable_frames_in_a_row + 1 - } else { - 0 - }; - if is_delete { - entry.refused_deletes_in_a_row += 1; - } else if !is_unreadable { - entry.refused_deletes_in_a_row = 0; - } - if is_unreadable { - entry.undecodable_frames_in_a_row - } else { - entry.refused_deletes_in_a_row - } - }); - let rejection = match decoded { - Ok(message) => self.deliver_application_message( - key, - frame.sequence, - message, - now_unix, - now_monotonic, - )?, - Err(error) => Some(CultMeshRudpApplicationRejection { - session: key, - operation: CultMeshRudpApplicationOperation::Unreadable, - message_id: String::new(), - record_key: None, - reason: CultMeshRudpRejectionReason::MessageUnreadable(format!("{error:#}")), - }), + ) { + Ok(message) => message, + Err(_) => continue, }; - if let Some(rejection) = rejection { - // A refused delete or unreadable frame below the cap is answered on the live session, - // which goes on: the peer keeps the connection a stray frame did not break. - if (is_unreadable || is_delete) - && run.is_some_and(|run| run < MAX_CONSECUTIVE_REFUSED_FRAMES) - { - let refusal = rejection.refusal_message(); - let packets = match self.sessions.remove(&key) { - Some(mut entry) => { - let packets = self.refusal_packets(&mut entry.session, &refusal, false); - self.sessions.insert(key, entry); - packets - } - None => Vec::new(), - }; - for packet in &packets { - if let Some(error) = self.send_packet(key.remote_addr, packet)? { - unsendable = Some(error); - break; - } - } - kept_rejection = Some(rejection); - continue; - } - // The capped frame is answered reliably, so a lost datagram is resent; the session ends once - // the peer has acknowledged the answer or the grace runs out (`end_answered_sessions`). When - // the answer cannot be queued, the session ends at once with the refusal as before. - if (is_unreadable || is_delete) - && run.is_some() - && !matches!( - rejection.reason, - CultMeshRudpRejectionReason::ResponseSendFailed(_) - ) - { - let refusal = rejection.refusal_message(); - let packets = match self.sessions.remove(&key) { - Some(mut entry) => { - let packets = self.refusal_packets(&mut entry.session, &refusal, true); - if !packets.is_empty() { - entry.ending_deadline_monotonic_millis = - Some(now_monotonic.saturating_add(CAPPED_ANSWER_GRACE_MILLIS)); - } - self.sessions.insert(key, entry); - packets - } - None => Vec::new(), - }; - if packets.is_empty() { - self.end_session(key, SessionEnd::Rejected(refusal))?; - return Ok(CultMeshRudpPollOutcome::ApplicationRejected(rejection)); - } - for mut packet in packets { - // Acknowledges nothing, as every refusal: the refused frame was not admitted. - packet.ack = 0; - packet.ack_mask = 0; - if let Some(error) = self.send_packet(key.remote_addr, &packet)? { - unsendable = Some(error); - break; - } - } - kept_rejection = Some(rejection); - break; - } + if let Some(rejection) = self.deliver_application_message( + key, + frame.sequence, + message, + now_unix, + now_monotonic, + )? { // The only reply `receive` returns is a Pong, which delivers no // frame, so no earlier send of this poll can precede a rejection. // A response that could not be sent ended its session where the @@ -758,11 +597,7 @@ where } } - let answering_capped_frame = self - .sessions - .get(&key) - .is_some_and(|entry| entry.ending_deadline_monotonic_millis.is_some()); - if packet.reliable && unsendable.is_none() && !answering_capped_frame { + if packet.reliable && unsendable.is_none() { let ack = self .sessions .get_mut(&key) @@ -774,11 +609,7 @@ where if let Some(error) = unsendable { self.end_session(key, SessionEnd::Unsendable(error))?; } - self.end_answered_sessions()?; - Ok(match kept_rejection { - Some(rejection) => CultMeshRudpPollOutcome::ApplicationRejected(rejection), - None => CultMeshRudpPollOutcome::Handled, - }) + Ok(CultMeshRudpPollOutcome::Handled) } /// Expire idle sessions and resend reliable packets whose deadline passed. @@ -792,16 +623,10 @@ where && now.saturating_sub(entry.created_at_monotonic_millis) <= lifetime_ms }); let expired = before - self.sessions.len(); - self.end_answered_sessions()?; let mut resends = Vec::new(); for (key, entry) in &mut self.sessions { - for mut packet in entry.session.due_resends(now) { - if entry.ending_deadline_monotonic_millis.is_some() { - // The resent answer to a capped frame acknowledges nothing either. - packet.ack = 0; - packet.ack_mask = 0; - } + for packet in entry.session.due_resends(now) { resends.push((*key, packet)); } } @@ -820,26 +645,6 @@ where }) } - /// Ends each session whose capped frame was answered once the peer has acknowledged every answer (a lost - /// answer is resent first), or when the grace has run out. - fn end_answered_sessions(&mut self) -> Result<()> { - let now = self.clock.now_monotonic_millis(); - let due: Vec = self - .sessions - .iter() - .filter(|(_, entry)| { - entry.ending_deadline_monotonic_millis.is_some_and(|deadline| { - entry.session.outstanding_reliable_packet_count() == 0 || now >= deadline - }) - }) - .map(|(key, _)| *key) - .collect(); - for key in due { - self.end_session(key, SessionEnd::Refused)?; - } - Ok(()) - } - /// `Ok(false)` when the Connect was refused because the session cap is full; /// only a socket failure is an `Err`. fn accept_connection( @@ -889,9 +694,6 @@ where created_at_monotonic_millis: now, last_activity_monotonic_millis: now, admitted_payload_bytes: 0, - undecodable_frames_in_a_row: 0, - refused_deletes_in_a_row: 0, - ending_deadline_monotonic_millis: None, }, ); accept @@ -1096,19 +898,6 @@ where } } } - CultNetMessage::DocumentDelete { - message_id, - record_key, - .. - } => { - return Ok(Some(CultMeshRudpApplicationRejection { - session: key, - operation: CultMeshRudpApplicationOperation::DocumentDelete, - message_id, - record_key: Some(record_key), - reason: CultMeshRudpRejectionReason::DeleteNotSupported, - })); - } _ => {} } Ok(None) @@ -1131,7 +920,7 @@ where }; let (refusal, mut unsendable) = match end { SessionEnd::Refused => (Vec::new(), None), - SessionEnd::Rejected(refusal) => (self.refusal_packets(&mut entry.session, &refusal, false), None), + SessionEnd::Rejected(refusal) => (self.refusal_packets(&mut entry.session, &refusal), None), SessionEnd::Unsendable(error) => (Vec::new(), Some(error)), }; entry.session.reset_peer_state(); @@ -1154,12 +943,11 @@ where Ok(()) } - /// The refusal a rejected peer is sent: unreliable and unordered, or (`reliable`) queued like any response. + /// The refusal a rejected peer is sent, as unreliable, unordered packets. fn refusal_packets( &self, session: &mut CultNetRudpSession, refusal: &CultNetMessage, - reliable: bool, ) -> Vec { encode_cultnet_message_to_vec(refusal, CultNetWireContract::CultNetSchemaV0) .and_then(|payload| { @@ -1167,8 +955,8 @@ where "schema", payload, CultNetRudpSendOptions { - reliable, - ordered: reliable, + reliable: false, + ordered: false, sequenced: false, now_ms: self.clock.now_monotonic_millis(), reliable_expire_after_ms: None, @@ -1341,9 +1129,6 @@ mod tests { // (`ForRemoteWriteRefused(request for "some:key", "not granted")`) and pinned by C#, TS and Python too. const WRITE_NOT_AUTHORIZED_HEX: &str = "85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72D92177726974655F6E6F745F617574686F72697A65643A206E6F74206772616E746564AB726F7574696E6748696E74C0A4636F6465B477726974655F6E6F745F617574686F72697A6564A764657461696C7384A56669656C64A97265636F72644B6579A576616C7565A8736F6D653A6B6579A461734F66C0A763757272656E74C0"; - // The same file's `message_invalid` vector: field `document`, reason "Raw put messages must contain a document map." - const MESSAGE_INVALID_HEX: &str = "85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72D93E6D6573736167655F696E76616C69643A2052617720707574206D65737361676573206D75737420636F6E7461696E206120646F63756D656E74206D61702EAB726F7574696E6748696E74C0A4636F6465AF6D6573736167655F696E76616C6964A764657461696C7384A56669656C64A8646F63756D656E74A576616C7565C0A461734F66C0A763757272656E74C0"; - fn hex(bytes: &[u8]) -> String { bytes.iter().map(|byte| format!("{byte:02X}")).collect() } @@ -1357,14 +1142,4 @@ mod tests { .unwrap(); assert_eq!(hex(&bytes), WRITE_NOT_AUTHORIZED_HEX); } - - #[test] - fn the_unreadable_refusal_is_the_pinned_message_invalid_bytes() { - let bytes = encode_cultnet_message_to_vec( - &message_invalid_message("document", "Raw put messages must contain a document map."), - CultNetWireContract::CultNetSchemaV0, - ) - .unwrap(); - assert_eq!(hex(&bytes), MESSAGE_INVALID_HEX); - } } diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index 0e3cba66..8d0c864c 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -343,17 +343,14 @@ fn refusal_seen_by( client: &mut CultNetRudpSocketTransportConnection, receipt: &CultNetRudpReliableSendReceipt, ) -> Result { - let (error, code, details) = typed_refusal_seen_by(client, receipt, None)?; + let (error, code, details) = typed_refusal_seen_by(client, receipt)?; assert_eq!((code, details), (None, None)); Ok(error) } -/// `server` is polled while the goodbye is awaited when the refusal ends its session only once the peer has -/// acknowledged it (the capped frame's answer). fn typed_refusal_seen_by( client: &mut CultNetRudpSocketTransportConnection, receipt: &CultNetRudpReliableSendReceipt, - mut server: Option<&mut Server>, ) -> Result<(String, Option, Option)> { let mut refusal = None; for _ in 0..2_000 { @@ -384,9 +381,6 @@ fn typed_refusal_seen_by( "a refusal must not acknowledge the refused message" ); for _ in 0..2_000 { - if let Some(server) = server.as_deref_mut() { - server.poll_once()?; - } client.receive_once()?; if !client.connected() { break; @@ -687,7 +681,7 @@ fn application_rejection_is_nonfatal_peer_scoped_refused_to_the_peer_and_unackno // The caller keeps the sink's text; the peer is sent fixed text, never it. // A refused put is typed write_not_authorized and names the record; nothing of it reached the sink. assert_eq!( - typed_refusal_seen_by(&mut publisher, &publish_receipt, None)?, + typed_refusal_seen_by(&mut publisher, &publish_receipt)?, ( "write_not_authorized: the catalog refused the document".to_string(), Some(CultNetErrorCode::WriteNotAuthorized), @@ -757,173 +751,6 @@ fn application_rejection_is_nonfatal_peer_scoped_refused_to_the_peer_and_unackno Ok(()) } -/// Sends `payload` reliably, polls until the server refuses it, and returns the typed refusal the -/// peer is sent. The session is not asserted here: the caller says whether it lives. -fn refusal_on_live_session( - server: &mut CultMeshRudpDocumentServer, - client: &mut CultNetRudpSocketTransportConnection, - payload: Vec, -) -> Result<(Option, Option, String)> { - client.send_reliable("schema", payload)?; - let mut rejected = false; - for _ in 0..2_000 { - if !rejected { - rejected = matches!(server.poll_once()?, CultMeshRudpPollOutcome::ApplicationRejected(_)); - } - if let Some(frame) = client.receive_once()? { - if let CultNetMessage::Error { error, code, details } = decode_cultnet_message_from_slice( - &frame.payload, - CultNetWireContract::CultNetSchemaV0, - )? { - assert!(rejected, "the refusal preceded the server's own outcome"); - return Ok((code, details, error)); - } - } - thread::sleep(Duration::from_millis(1)); - } - anyhow::bail!("no typed refusal reached the peer") -} - -fn garbage() -> Vec { - let mut payload = vec![0xAE]; - payload.extend_from_slice(b"CANARY-garbage"); - payload -} - -fn delete_of(key: &str) -> Vec { - encode_cultnet_message_to_vec( - &CultNetMessage::DocumentDelete { - message_id: "del".into(), - schema_id: "sha256:note".into(), - record_key: key.into(), - }, - CultNetWireContract::CultNetSchemaV0, - ) - .unwrap() -} - -fn valid_put(key: &str) -> Vec { - encode_cultnet_message_to_vec( - &CultNetMessage::DocumentPutRaw { - message_id: format!("put-{key}"), - document: document(key, vec![1, 2, 3]), - }, - CultNetWireContract::CultNetSchemaV0, - ) - .unwrap() -} - -fn stored_after_poll( - server: &mut CultMeshRudpDocumentServer, - client: &mut CultNetRudpSocketTransportConnection, - sink: &Sink, - receipts: usize, -) -> Result<()> { - client.send_reliable("schema", valid_put(&format!("k{receipts}")))?; - for _ in 0..2_000 { - server.poll_once()?; - client.receive_once()?; - if sink.0.lock().unwrap().receipts.len() >= receipts { - return Ok(()); - } - thread::sleep(Duration::from_millis(1)); - } - anyhow::bail!("a valid put on the kept session was not stored") -} - -#[test] -fn an_unreadable_frame_is_refused_typed_and_the_session_kept_until_eight_in_a_row() -> Result<()> { - let sink = Sink::default(); - let mut server = server(Default::default(), Clock::new(49_000), sink.clone(), Source::default())?; - let target = server.local_addr()?; - let mut garbler = client(target, 111)?; - let mut survivor = client(target, 113)?; - connect(&mut server, &mut [&mut garbler, &mut survivor])?; - let message_invalid = ( - Some(CultNetErrorCode::MessageInvalid), - Some(CultNetErrorDetails { - field: Some("message".into()), - value: None, - as_of: None, - current: None, - }), - "message_invalid: the message could not be read".to_string(), - ); - - // Seven in a row are answered and the session lives, a valid frame resets the run, and seven more - // are answered again: a counter that never resets, or a cap below eight, ends the session early. - for _ in 0..7 { - assert_eq!(refusal_on_live_session(&mut server, &mut garbler, garbage())?, message_invalid); - assert!(garbler.connected()); - } - stored_after_poll(&mut server, &mut garbler, &sink, 1)?; - for _ in 0..7 { - assert_eq!(refusal_on_live_session(&mut server, &mut garbler, garbage())?, message_invalid); - assert!(garbler.connected()); - } - assert_eq!(server.session_count(), 2); - - // The eighth consecutive one is answered too, then the session ends with a goodbye. - let receipt = garbler.send_reliable("schema", garbage())?; - let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = server.poll_once()? else { - panic!("the eighth consecutive unreadable frame must be refused"); - }; - assert_eq!(rejection.operation, CultMeshRudpApplicationOperation::Unreadable); - let (error, code, _) = typed_refusal_seen_by(&mut garbler, &receipt, Some(&mut server))?; - assert_eq!((error.as_str(), code), (message_invalid.2.as_str(), message_invalid.0)); - drain_until_idle(&mut server)?; - assert_eq!(server.session_count(), 1, "only the capped peer's session ended"); - assert_eq!(sink.0.lock().unwrap().receipts.len(), 1); - Ok(()) -} - -#[test] -fn a_refused_delete_keeps_the_session_until_eight_in_a_row() -> Result<()> { - let sink = Sink::default(); - let mut server = server(Default::default(), Clock::new(50_000), sink.clone(), Source::default())?; - let target = server.local_addr()?; - let mut deleter = client(target, 112)?; - connect(&mut server, &mut [&mut deleter])?; - let refused = |key: &str| { - ( - Some(CultNetErrorCode::WriteNotAuthorized), - Some(CultNetErrorDetails { - field: Some("recordKey".into()), - value: Some(key.into()), - as_of: None, - current: None, - }), - "write_not_authorized: the server does not take deletes".to_string(), - ) - }; - for _ in 0..7 { - assert_eq!(refusal_on_live_session(&mut server, &mut deleter, delete_of("note:victim"))?, refused("note:victim")); - assert!(deleter.connected()); - } - // A decoded non-delete frame resets the run. - stored_after_poll(&mut server, &mut deleter, &sink, 1)?; - for _ in 0..7 { - assert_eq!(refusal_on_live_session(&mut server, &mut deleter, delete_of("note:victim"))?, refused("note:victim")); - } - let receipt = deleter.send_reliable("schema", delete_of("note:victim"))?; - let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = server.poll_once()? else { - panic!("the eighth consecutive delete must be refused"); - }; - assert_eq!(rejection.operation, CultMeshRudpApplicationOperation::DocumentDelete); - assert_eq!(rejection.reason, CultMeshRudpRejectionReason::DeleteNotSupported); - let (error, code, details) = typed_refusal_seen_by(&mut deleter, &receipt, Some(&mut server))?; - assert_eq!( - (error, code, details), - { - let (c, d, e) = refused("note:victim"); - (e, c, d) - } - ); - drain_until_idle(&mut server)?; - assert_eq!(server.session_count(), 0); - Ok(()) -} - #[test] fn snapshot_response_resends_until_acknowledged() -> Result<()> { let clock = Clock::new(51_000); @@ -1629,11 +1456,6 @@ impl RawPeer { /// Sends `message` reliably and in order; returns its sequences. fn send(&mut self, message: &CultNetMessage) -> Result> { let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?; - self.send_payload(payload) - } - - /// Sends `payload` as one reliable, ordered schema frame; returns its sequences. - fn send_payload(&mut self, payload: Vec) -> Result> { let options = cultnet_rs::CultNetRudpSendOptions { reliable: true, ordered: true, @@ -1921,138 +1743,3 @@ fn a_response_that_is_too_large_to_send_is_named_emsgsize() -> Result<()> { ); Ok(()) } - -/// A peer on `server` that has sent seven unreadable frames, each answered and processed, and then the -/// eighth, whose answer is waiting at its socket. Returns the peer and the eighth frame's sequence. -fn peer_at_the_cap(server: &mut Server, connection_id: u32) -> Result<(RawPeer, u32)> { - let mut peer = RawPeer::connect(server, connection_id)?; - for _ in 0..7 { - peer.send_payload(garbage())?; - assert!(matches!( - server.poll_once()?, - CultMeshRudpPollOutcome::ApplicationRejected(_) - )); - for packet in peer.drain() { - peer.session.receive(&packet, 3)?; - } - } - let capped = peer.send_payload(garbage())?[0]; - assert!(matches!( - server.poll_once()?, - CultMeshRudpPollOutcome::ApplicationRejected(_) - )); - Ok((peer, capped)) -} - -fn is_goodbye(packet: &cultnet_rs::CultNetRudpPacket) -> bool { - packet.packet_type == cultnet_rs::CultNetRudpPacketType::Disconnect -} - -fn is_answer(packet: &cultnet_rs::CultNetRudpPacket) -> bool { - packet.packet_type == cultnet_rs::CultNetRudpPacketType::Data && packet.reliable -} - -/// The capped frame's answer is reliable: its datagram is lost, the server resends it, and only once the -/// peer has acknowledged it does the goodbye follow. Neither transmission acknowledges the capped frame. -#[test] -fn a_lost_capped_answer_is_resent_and_the_goodbye_waits_for_its_acknowledgement() -> Result<()> { - let clock = Clock::new(81_000); - let mut server = server(Default::default(), clock.clone(), Sink::default(), Source::default())?; - let (mut peer, capped) = peer_at_the_cap(&mut server, 81)?; - - let first = peer.drain(); - assert!(first.iter().any(is_answer), "fixture: the answer was sent, and is lost here"); - assert!(!first.iter().any(is_goodbye), "no goodbye before the answer is acknowledged"); - assert!( - !first.iter().any(|packet| acknowledges(packet, capped)), - "nothing sent for the capped frame acknowledges it" - ); - assert_eq!(server.session_count(), 1); - - clock.set_monotonic(81_000 + 60); - server.maintain()?; - let resent = peer.drain(); - let answer = resent.iter().find(|packet| is_answer(packet)).expect("the lost answer is resent"); - assert!(!acknowledges(answer, capped), "the resent answer acknowledges nothing the peer sent"); - assert!(!resent.iter().any(is_goodbye), "still no goodbye"); - let delivered = peer.session.receive(answer, 4)?.delivered; - let Ok(CultNetMessage::Error { code, .. }) = - decode_cultnet_message_from_slice(&delivered[0].payload, CultNetWireContract::CultNetSchemaV0) - else { - panic!("the resent answer is a typed refusal"); - }; - assert_eq!(code, Some(CultNetErrorCode::MessageInvalid)); - - peer.send_packet(&peer.session.create_ack_for(answer.sequence))?; - assert_eq!(server.poll_once()?, CultMeshRudpPollOutcome::Handled); - assert!(peer.drain().iter().any(is_goodbye), "acknowledged: the goodbye follows"); - assert_eq!(server.session_count(), 0); - Ok(()) -} - -/// A peer that never acknowledges the answer is not kept for ever: the session ends after the grace. -#[test] -fn a_capped_session_whose_answer_is_never_acknowledged_ends_after_the_grace() -> Result<()> { - let clock = Clock::new(82_000); - let mut server = server(Default::default(), clock.clone(), Sink::default(), Source::default())?; - let (peer, _) = peer_at_the_cap(&mut server, 82)?; - peer.drain(); - - clock.set_monotonic(82_000 + 1_999); - server.maintain()?; - assert!(!peer.drain().iter().any(is_goodbye), "inside the grace"); - assert_eq!(server.session_count(), 1); - clock.set_monotonic(82_000 + 2_000); - server.maintain()?; - assert!(peer.drain().iter().any(is_goodbye), "the grace ran out"); - assert_eq!(server.session_count(), 0); - Ok(()) -} - -/// Nothing behind the capped frame is served while its answer is delivered. -#[test] -fn nothing_behind_the_capped_frame_is_served() -> Result<()> { - let sink = Sink::default(); - let mut server = server(Default::default(), Clock::new(83_000), sink.clone(), Source::default())?; - let (mut peer, _) = peer_at_the_cap(&mut server, 83)?; - peer.send_payload(valid_put("behind-the-cap"))?; - for _ in 0..3 { - server.poll_once()?; - } - assert!(sink.0.lock().unwrap().receipts.is_empty(), "the put behind the cap was not stored"); - Ok(()) -} - -/// A Ping inside the grace is answered with a Pong that acknowledges nothing: its ack fields are built from -/// what the session received, which includes the capped frame. -#[test] -fn a_pong_in_the_grace_after_the_cap_acknowledges_nothing() -> Result<()> { - let mut server = server(Default::default(), Clock::new(84_000), Sink::default(), Source::default())?; - let (mut peer, capped) = peer_at_the_cap(&mut server, 84)?; - peer.drain(); - - let ping = peer.session.create_ping(b"alive".to_vec()); - peer.send_packet(&ping)?; - assert_eq!(server.poll_once()?, CultMeshRudpPollOutcome::Handled); - let pong = peer - .drain() - .into_iter() - .find(|packet| packet.packet_type == cultnet_rs::CultNetRudpPacketType::Pong) - .expect("the ping is answered"); - assert_eq!(pong.payload, b"alive"); - assert!(!acknowledges(&pong, capped), "the Pong acknowledges the capped frame"); - assert_eq!((pong.ack, pong.ack_mask), (0, 0)); - Ok(()) -} - -/// The grace deadline saturates: at the end of the clock the answer is not given up on at once. -#[test] -fn the_grace_deadline_does_not_wrap_at_the_end_of_the_clock() -> Result<()> { - let mut server = server(Default::default(), Clock::new(u64::MAX - 10), Sink::default(), Source::default())?; - let (peer, _) = peer_at_the_cap(&mut server, 85)?; - peer.drain(); - server.maintain()?; - assert!(!peer.drain().iter().any(is_goodbye), "the deadline wrapped and the session ended before its grace"); - assert_eq!(server.session_count(), 1); - Ok(()) -} diff --git a/packages/cultmesh-ts/CHANGELOG.md b/packages/cultmesh-ts/CHANGELOG.md index d06e753b..22d99904 100644 --- a/packages/cultmesh-ts/CHANGELOG.md +++ b/packages/cultmesh-ts/CHANGELOG.md @@ -16,7 +16,6 @@ All notable changes to `@gamecult/cultmesh-ts` are recorded here, following ### Added -- `createRudpDocumentServer` ends a session after `CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES` (8) consecutive undecodable frames or consecutive deletes the write authority refuses; fewer are answered and the session is kept. Any decoded frame resets the undecodable run, any decoded non-refused-delete frame resets the delete run. The same cap holds in every runtime. - `CultMeshCatalogValue`, exported. - `documentById` and `tryDocumentById` on `CultMeshDocumentCatalog`, which address one document by its document id when a catalog holds several documents of one type. diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index f065dd83..e8d526cc 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -4121,16 +4121,6 @@ export interface CultMeshRudpDocumentServerOptions extends CultMeshRudpSocketOpt /** A write authority that has not answered in this long has refused the write. */ export const CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_DECIDE_TIMEOUT_MS = 5_000; -/** - * How many consecutive refused frames of one kind a `createRudpDocumentServer` session may send before it is - * ended: undecodable frames (reset by any frame that decodes) and deletes the write authority refuses (reset by any - * decoded frame that is not a refused delete). Fewer are answered with a typed refusal and the session is kept; the - * frame that reaches this number is answered too, then the session ends. The same number is the cap in every runtime. - */ -export const CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES = 8; -/** How long a session whose capped frame was answered waits for the peer to acknowledge the answers before it ends anyway. */ -const CULTMESH_RUDP_DOCUMENT_SERVER_CAPPED_ANSWER_GRACE_MS = 2000; - /** Replies larger than one fragment are split into fragments of at most this many bytes. */ export const CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_MAX_FRAGMENT_BYTES = 1200; @@ -4183,29 +4173,7 @@ export interface CultMeshRudpDocumentServer { close(): void; } -/** - * A write the peer refused: the typed `cultnet.error.v0` it answered with (`write_not_authorized`, - * `message_invalid`, ...), whichever of the transport ack and the result frame came first. - */ -export class CultMeshRudpWriteRefusedError extends Error { - readonly code: string | undefined; - readonly response: Extract; - constructor(response: Extract) { - super(response.error); - this.name = "CultMeshRudpWriteRefusedError"; - this.code = response.code ?? undefined; - this.response = response; - } -} - export interface CultMeshRudpDocumentPublishOptions extends CultMeshRudpPeerOptions { - /** - * How long after the put's transport ack a refusal may still arrive. A reliable-transport ack is not - * admission; the outcome is the peer's result frame. A server that answers a put only on refusal sends no - * success frame, so the put is reported published once this window passes with no refusal; a success reply - * ends the wait early. Default 500 ms. - */ - resultWindowMs?: number; messageId?: string; sourceRuntimeId?: string; sourceAgentId?: string; @@ -5569,11 +5537,6 @@ export class CultMesh { closed: boolean; /** A packet it owed could never be sent, so the session ended. */ unsendable: boolean; - /** Consecutive undecodable frames, and consecutive deletes the authority refused. */ - unreadableInARow: number; - refusedDeletesInARow: number; - /** Set once the peer's capped frame is answered: the session ends when the answers are acknowledged, or at this time. */ - endingDeadlineMs?: number; work: Promise; }; const sessions = new Map(); @@ -5628,8 +5591,6 @@ export class CultMesh { connectPayload, closed: false, unsendable: false, - unreadableInARow: 0, - refusedDeletesInARow: 0, session: new CultNetRudpSession({ connectionId, initialSequence: options.initialSequence, @@ -5694,8 +5655,7 @@ export class CultMesh { record: SessionRecord, payload: Uint8Array, ): Promise { - // Nothing after the capped frame is served while its answer is delivered. - if (record.closed || record.endingDeadlineMs !== undefined) return; + if (record.closed) return; let decoded: unknown; let message: CultNetMessage; try { @@ -5705,21 +5665,12 @@ export class CultMesh { // A frame the server cannot read is judged as the write it looks like (a put when it does not even decode), so // an ungranted peer hears the same refusal for it as for any other write, not a parse error. const write = readableWrite(decoded); - record.unreadableInARow += 1; - if (decoded !== undefined && !write) { - // It decodes but is no message the parser takes (an unknown schemaVersion, a non-map) and is no write the - // authority could judge: answered and counted like any other unreadable frame. - reportError(error); - sendSchemaMessage(record, cultNetMessageInvalidError("message", "CultMesh RUDP document server could not read the message.")); - } else if (!await writeRefused(record, write?.operation ?? "put", write?.schemaId ?? "", write?.recordKey ?? "", null, null)) { - reportError(error); - sendSchemaMessage(record, cultNetMessageInvalidError("message", "CultMesh RUDP document server could not read the write.")); - } - if (record.unreadableInARow >= CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES) endCappedSession(record); + if (decoded !== undefined && !write) throw error; + if (await writeRefused(record, write?.operation ?? "put", write?.schemaId ?? "", write?.recordKey ?? "", null, null)) return; + reportError(error); + sendSchemaMessage(record, cultNetMessageInvalidError("message", "CultMesh RUDP document server could not read the write.")); return; } - record.unreadableInARow = 0; - if (message.schemaVersion !== "cultnet.document_delete.v0") record.refusedDeletesInARow = 0; switch (message.schemaVersion) { case "cultnet.snapshot_request.v0": { if (!options.getCache) { @@ -5793,12 +5744,7 @@ export class CultMesh { } return; case "cultnet.document_delete.v0": { - if (await writeRefused(record, "delete", message.schemaId, message.recordKey, null, null)) { - record.refusedDeletesInARow += 1; - if (record.refusedDeletesInARow >= CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES) endCappedSession(record); - return; - } - record.refusedDeletesInARow = 0; + if (await writeRefused(record, "delete", message.schemaId, message.recordKey, null, null)) return; if (!options.onDocumentDeleteRaw) { sendSchemaMessage(record, cultNetTextError("CultMesh RUDP document server has no delete handler.")); return; @@ -5949,17 +5895,6 @@ export class CultMesh { sendRudpDatagram(socket, goodbye, record.remote.port, record.remote.address, () => {}); } - /** - * A session that reached the cap of consecutive refused frames: its last refusal went out like any reply - * (reliable, so a lost datagram is resent), and the session ends once the peer has acknowledged every reply or - * the grace runs out. The resend timer ends it. - */ - function endCappedSession(record: SessionRecord): void { - const key = `${record.remote.address}:${record.remote.port}`; - if (sessions.get(key) !== record) return; - record.endingDeadlineMs ??= Date.now() + CULTMESH_RUDP_DOCUMENT_SERVER_CAPPED_ANSWER_GRACE_MS; - } - /** The session owes its peer a packet that can never be sent as built. */ function endUnsendableSession(record: SessionRecord, error: Error): void { if (record.unsendable) return; @@ -6093,11 +6028,6 @@ export class CultMesh { closeSession(record); continue; } - if (record.endingDeadlineMs !== undefined - && (record.session.outstandingReliablePacketCount === 0 || nowMs >= record.endingDeadlineMs)) { - endRefusedSession(key, record); - continue; - } for (const packet of record.session.dueResends(nowMs)) { sendPacket(record, packet); } @@ -6179,23 +6109,11 @@ export class CultMesh { requireNonEmpty(runtimeId, "runtimeId"); requireNonEmpty(recordKey, "recordKey"); const registry = new CultNetDocumentRegistry([binding]); - const messageId = options.messageId ?? `${runtimeId}:${binding.definition.type}:${recordKey}`; const peer = await CultMesh.createRudpPeer(runtimeId, connectionId, endpoint, options); try { - // A transport ack is not admission: the put completes on its result frame. A refusal rejects with the - // typed error whether it arrives before the ack, after it, or inside the result window. - let refusal: CultMeshRudpWriteRefusedError | undefined; - let succeeded = false; - let wake: (() => void) | undefined; - peer.on("message", (message: CultNetMessage) => { - if (message.schemaVersion === "cultnet.error.v0") refusal ??= new CultMeshRudpWriteRefusedError(message); - else if ("messageId" in message && message.messageId === messageId) succeeded = true; - else return; - wake?.(); - }); peer.send(registry.createRawDocumentPutMessage( binding, - messageId, + options.messageId ?? `${runtimeId}:${binding.definition.type}:${recordKey}`, recordKey, value, { @@ -6206,13 +6124,6 @@ export class CultMesh { }, )); await peer.flush(options.flushTimeoutMs ?? 30_000); - if (!refusal && !succeeded) { - await new Promise((resolve) => { - const timer = setTimeout(resolve, options.resultWindowMs ?? 500); - wake = () => { clearTimeout(timer); resolve(); }; - }); - } - if (refusal) throw refusal; } finally { peer.close(); } @@ -6231,7 +6142,6 @@ export class CultMesh { const receipt = new Promise>((resolve, reject) => { const timeout = setTimeout(() => { cleanup(); reject(new Error(`Timed out waiting for RUDP application receipt for ${recordKey}.`)); }, options.receiptTimeoutMs ?? 5_000); const onMessage = (message: CultNetMessage): void => { - if (message.schemaVersion === "cultnet.error.v0") { cleanup(); reject(new CultMeshRudpWriteRefusedError(message)); return; } if (message.schemaVersion !== "cultnet.document_put_raw.v0" || message.messageId !== messageId || message.document.schemaId !== receiptBinding.definition.schemaId) return; try { const decoded = decodeDocumentPayload(receiptBinding.definition, message.document.payload); diff --git a/packages/cultmesh-ts/test/client-write-completion.test.ts b/packages/cultmesh-ts/test/client-write-completion.test.ts deleted file mode 100644 index 8dd122ed..00000000 --- a/packages/cultmesh-ts/test/client-write-completion.test.ts +++ /dev/null @@ -1,117 +0,0 @@ -// Heimdall's invariant: no client may treat a reliable-transport ack as write admission. A put completes only -// on its result frame (write_not_authorized, or the success reply). Each test serves a client a peer that -// acknowledges the put and refuses it, the ack arriving before the refusal and after it. -import assert from "node:assert/strict"; -import { createSocket, type Socket } from "node:dgram"; -import test from "node:test"; -import { z } from "zod"; -import { defineDocumentType } from "@gamecult/cultcache-ts"; -import { - CultNetDocumentRegistry, - CultNetRudpSession, - cultNetWriteNotAuthorizedError, - decodeRudpPacket, - defineCultNetDocumentBinding, - encodeCultNetMessageForWire, - encodeRudpPacket, -} from "@gamecult/cultnet-ts"; -import { encodeWireValue } from "@gamecult/cultnet-ts/wire"; -import { CultMesh, CultMeshGrantAllWrites, CultMeshRudpWriteRefusedError } from "../src/index"; - -const isTheRefusal = (error: unknown): boolean => - error instanceof CultMeshRudpWriteRefusedError && error.code === "write_not_authorized"; - -const note = defineDocumentType({ - type: "cultmesh.completion_note", - schemaId: "cultmesh.completion_note.v0", - schema: z.object({ noteId: z.string(), text: z.string() }), - name: "noteId", -}); -const binding = defineCultNetDocumentBinding({ definition: note }); -const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); - -type Order = "ack-then-refusal" | "refusal-then-ack"; - -/** A peer that acknowledges every reliable put and refuses it with write_not_authorized, in the given order. */ -async function refusingPeer(connectionId: number, order: Order) { - const socket: Socket = createSocket("udp4"); - await new Promise((resolve) => socket.bind(0, "127.0.0.1", resolve)); - const session = new CultNetRudpSession({ connectionId, resendDelayMs: 50 }); - const timers = new Set(); - let closed = false; - const send = (packet: Parameters[0], port: number, address: string) => { - if (!closed) socket.send(encodeRudpPacket(packet), port, address); - }; - const later = (action: () => void) => { const timer = setTimeout(() => { timers.delete(timer); action(); }, 120); timers.add(timer); }; - socket.on("message", (wire, remote) => { - const packet = decodeRudpPacket(wire); - const now = Date.now(); - if (packet.packetType === "connect") { - send(session.acceptConnect(packet, now), remote.port, remote.address); - return; - } - const result = session.receive(packet, now); - if (result.reply) send(result.reply, remote.port, remote.address); - for (const frame of result.delivered) { - if (frame.channelId !== "schema") continue; - const ack = session.createAckForReceived(packet.sequence); - const refusal = session.sendMany( - "schema", - encodeWireValue(encodeCultNetMessageForWire(cultNetWriteNotAuthorizedError("tests:completion:a"), "cultnet.schema.v0")), - { reliable: true, ordered: true, nowMs: now }, - ); - if (order === "ack-then-refusal") { - send(ack, remote.port, remote.address); - later(() => refusal.forEach((p) => send(p, remote.port, remote.address))); - } else { - // The refusal carries no acknowledgement of its own; the ack follows it. - refusal.forEach((p) => send({ ...p, ack: 0, ackMask: 0 }, remote.port, remote.address)); - later(() => send(ack, remote.port, remote.address)); - } - } - if (packet.reliable) send(session.createAckForReceived(packet.sequence), remote.port, remote.address); - }); - return { endpoint: `rudp://127.0.0.1:${socket.address().port}`, close: () => { closed = true; timers.forEach(clearTimeout); socket.close(); } }; -} - -const orders: Order[] = ["ack-then-refusal", "refusal-then-ack"]; -let nextConnection = 0x10205000; - -for (const order of orders) { - test(`publishRudpDocumentOnce does not complete on the ack of a refused put (${order})`, async () => { - const peer = await refusingPeer(nextConnection, order); - try { - await assert.rejects( - CultMesh.publishRudpDocumentOnce("completion-once", nextConnection++, peer.endpoint, binding, "tests:completion:a", { noteId: "a", text: "x" }, { resendDelayMs: 50, resendPollMs: 5, flushTimeoutMs: 2_000 }), - isTheRefusal, - ); - } finally { peer.close(); } - }); - - test(`publishRudpDocumentAndWaitForReceipt reports a refused put as refused, never as success (${order})`, async () => { - const peer = await refusingPeer(nextConnection, order); - try { - await assert.rejects( - CultMesh.publishRudpDocumentAndWaitForReceipt("completion-receipt", nextConnection++, peer.endpoint, binding, "tests:completion:a", { noteId: "a", text: "x" }, binding, { resendDelayMs: 50, resendPollMs: 5, receiptTimeoutMs: 800 }), - isTheRefusal, - ); - } finally { peer.close(); } - await sleep(10); - }); -} - -test("publishRudpDocumentOnce still publishes a put the server admitted", async () => { - const connectionId = nextConnection++; - const puts: string[] = []; - const server = CultMesh.createRudpDocumentServer("completion-admit", connectionId, { - documents: new CultNetDocumentRegistry([binding]), - writeAuthority: new CultMeshGrantAllWrites(), - bindHost: "127.0.0.1", bindPort: 0, resendDelayMs: 50, resendPollMs: 5, - onDocumentPutRaw: (put) => { puts.push(put.recordKey); }, - }); - await server.start(); - try { - await CultMesh.publishRudpDocumentOnce("completion-admit-client", connectionId, `rudp://127.0.0.1:${server.bind.port}`, binding, "tests:completion:a", { noteId: "a", text: "x" }, { resendDelayMs: 50, resendPollMs: 5, resultWindowMs: 200 }); - assert.deepEqual(puts, ["tests:completion:a"]); - } finally { server.close(); } -}); diff --git a/packages/cultmesh-ts/test/remote-write-authority.test.ts b/packages/cultmesh-ts/test/remote-write-authority.test.ts index 3c35171e..46fc5268 100644 --- a/packages/cultmesh-ts/test/remote-write-authority.test.ts +++ b/packages/cultmesh-ts/test/remote-write-authority.test.ts @@ -1,13 +1,11 @@ // A peer's put or delete is applied only when the server's write authority grants it. Ports the behaviours of // cultmesh-py test_remote_write_authority.py (no forwarding: this server does not forward non-primary writes). import assert from "node:assert/strict"; -import { createSocket, type Socket } from "node:dgram"; import { readFileSync } from "node:fs"; import { join } from "node:path"; import test from "node:test"; import { z } from "zod"; import { documentPayloadFromBytes, defineDocumentType } from "@gamecult/cultcache-ts"; -import { decodeRudpPacket } from "@gamecult/cultnet-ts"; import { encodeWireValue } from "@gamecult/cultnet-ts/wire"; import { CultNetDocumentRegistry, @@ -18,7 +16,6 @@ import { } from "@gamecult/cultnet-ts"; import { CultMesh, - CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES, CultMeshGrantAllWrites, type CultMeshRemoteWriteAuthority, type CultMeshRemoteWriteDecision, @@ -508,178 +505,3 @@ test("a granted peer's unreadable write reaches onError and gets a malformed ref } finally { r.close(); } } }); - -const CAP = 8; // the cap every runtime shares; the exported constant must stay this number - -test("the consecutive-refused-frames cap is the number every runtime shares", () => { - assert.equal(CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES, CAP); -}); - -test("unreadable frames are answered and the session kept until eight in a row; any decoded frame resets the run", async () => { - const r = await rawRig(new Recording(grant)); - try { - const batch = async (frames: number) => { - for (let i = 0; i < frames; i++) r.send(Uint8Array.from([0xc1])); - await sleep(400); - }; - await batch(CAP - 1); - assert.equal(r.replies.length, CAP - 1, "each is answered"); - r.send(encodeWireValue(put(KEY_A))); - await sleep(250); - assert.deepEqual(r.puts.map((p) => p.recordKey), [KEY_A], "the session was kept through seven"); - await batch(CAP - 1); - assert.equal(r.replies.length, 2 * (CAP - 1), "the valid frame reset the run"); - r.send(Uint8Array.from([0xc1])); - await sleep(300); - assert.equal(r.replies.length, 2 * (CAP - 1) + 1, "the eighth consecutive one is answered too"); - assert.ok(r.replies.every((reply) => (reply as { code?: string }).code === "message_invalid")); - // The goodbye reached the peer: its client no longer takes a send. - assert.throws(() => r.send(encodeWireValue(put(KEY_B))), /before the session is connected/, "then the session ended"); - assert.deepEqual(r.puts.map((p) => p.recordKey), [KEY_A]); - } finally { r.close(); } -}); - -test("deletes the authority refuses are answered and the session kept until eight in a row; any other decoded frame resets the run", async () => { - const authority = new Recording((request) => request.recordKey === KEY_B ? { granted: true } : { granted: false }); - await using(await rig(authority), async (r) => { - const refuse = async (deletes: number) => { - for (let i = 0; i < deletes; i++) r.peer.send(del(KEY_A)); - await sleep(400); - }; - await refuse(CAP - 1); - assert.equal(r.replies.length, CAP - 1); - r.peer.send(del(KEY_B)); // granted: resets the run - await sleep(250); - assert.deepEqual(r.deletes.map((d) => d.recordKey), [KEY_B]); - await refuse(CAP - 1); - assert.equal(r.replies.length, 2 * (CAP - 1)); - r.peer.send(put(KEY_B)); // a decoded non-delete resets too - await sleep(250); - await refuse(CAP - 1); - assert.equal(r.replies.length, 3 * (CAP - 1), "a put reset the run"); - r.peer.send(del(KEY_A)); - await sleep(300); - assert.equal(r.replies.length, 3 * (CAP - 1) + 1, "the eighth consecutive refused delete is answered too"); - for (const reply of r.replies) assertRefused(reply, KEY_A); - assert.throws(() => r.peer.send(del(KEY_B)), /before the session is connected/, "then the session ended"); - assert.equal(r.deletes.length, 1); - }); -}); - -test("a frame that decodes but is no message the parser takes is answered, counted, and ended at eight in a row like any unreadable frame", async () => { - const r = await rawRig(new Recording(grant)); - const unknown = () => encodeWireValue({ schemaVersion: "cultnet.not-a-message.v0", messageId: "x" }); - try { - const batch = async (frames: number) => { - for (let i = 0; i < frames; i++) r.send(unknown()); - await sleep(400); - }; - await batch(CAP - 1); - assert.equal(r.replies.length, CAP - 1, "each is answered"); - assert.equal(r.errors.length, CAP - 1, "and reported"); - r.send(encodeWireValue(put(KEY_A))); - await sleep(250); - assert.deepEqual(r.puts.map((p) => p.recordKey), [KEY_A], "the session was kept through seven"); - await batch(CAP - 1); - assert.equal(r.replies.length, 2 * (CAP - 1), "the decoded put reset the run"); - r.send(unknown()); - await sleep(300); - assert.equal(r.replies.length, 2 * (CAP - 1) + 1, "the eighth is answered too"); - for (const reply of r.replies) { - assert.equal((reply as { code?: string }).code, "message_invalid"); - assert.equal((reply as { details: { field: string } }).details.field, "message"); - } - assert.throws(() => r.send(unknown()), /before the session is connected/, "then the session ended"); - } finally { r.close(); } -}); - -/** A UDP relay between a client and a server that drops the first (or, with `dropEvery`, every) server datagram carrying `payloadContains` once armed. */ -async function lossyRelay(serverPort: number, payloadContains: string, dropEvery = false) { - const dropped: unknown[] = []; - let armed = false; - let clientAddress: { port: number; address: string } | undefined; - const toClient: Socket = createSocket("udp4"); - const toServer: Socket = createSocket("udp4"); - toServer.on("message", (wire) => { - if (!clientAddress) return; - if (armed && (dropEvery || dropped.length === 0)) { - const packet = decodeRudpPacket(wire); - if (packet.packetType === "data" && Buffer.from(packet.payload ?? []).includes(payloadContains)) { - dropped.push(packet); - return; - } - } - toClient.send(wire, clientAddress.port, clientAddress.address); - }); - toClient.on("message", (wire, remote) => { - clientAddress = remote; - toServer.send(wire, serverPort, "127.0.0.1"); - }); - await new Promise((resolve) => toClient.bind(0, "127.0.0.1", resolve)); - await new Promise((resolve) => toServer.bind(0, "127.0.0.1", resolve)); - return { - port: toClient.address().port, - dropped, - arm: () => { armed = true; }, - close: () => { toClient.close(); toServer.close(); }, - }; -} - -async function cappedOverRelay(dropEvery: boolean, body: (x: { relay: Awaited>; client: { send(channel: string, payload: Uint8Array): void }; replies: CultNetMessage[] }) => Promise) { - const connectionId = nextConnection++; - const server = CultMesh.createRudpDocumentServer("authority-lossy", connectionId, { - documents: new CultNetDocumentRegistry([defineCultNetDocumentBinding({ definition: note })]), - writeAuthority: new Recording(grant), bindHost: "127.0.0.1", bindPort: 0, resendDelayMs: 25, resendPollMs: 5, - onDocumentPutRaw: () => {}, onError: () => {}, - }); - await server.start(); - const relay = await lossyRelay(server.bind.port, "message_invalid", dropEvery); - const client = await CultMesh.createRudpClient("lossy", connectionId, `rudp://127.0.0.1:${relay.port}`, { resendDelayMs: 25, resendPollMs: 5 }); - try { - client.connect(); - const started = Date.now(); - while (!client.connected) { - if (Date.now() - started > 2_000) throw new Error("not connected"); - await sleep(5); - } - const replies: CultNetMessage[] = []; - new CultNetPeer(client, { wireContract: "cultnet.schema.v0" }).on("message", (m) => replies.push(m)); - for (let i = 0; i < CAP - 1; i++) client.send("schema", Uint8Array.from([0xc1])); - await sleep(300); - assert.equal(replies.length, CAP - 1); - relay.arm(); - client.send("schema", Uint8Array.from([0xc1])); - await body({ relay, client, replies }); - } finally { client.close(); server.close(); relay.close(); } -} - -test("the capped frame's answer is resent when its datagram is lost, and the goodbye follows only after it arrived", async () => { - await cappedOverRelay(false, async ({ relay, client, replies }) => { - await sleep(600); - assert.equal(relay.dropped.length, 1, "fixture: the first transmission of the last answer was lost"); - assert.equal(replies.length, CAP, "the lost answer was resent and arrived"); - assert.equal((replies[CAP - 1] as { code?: string }).code, "message_invalid"); - assert.throws(() => client.send("schema", Uint8Array.from([0xc1])), /before the session is connected/, "then the session ended"); - }); -}); - -test("a capped session whose answer is never acknowledged still ends, after the grace", async () => { - await cappedOverRelay(true, async ({ client, replies }) => { - await sleep(1200); - assert.equal(replies.length, CAP - 1, "fixture: the last answer never arrives"); - client.send("schema", Uint8Array.from([0xc1])); // still connected: the server is waiting for the acknowledgement - await sleep(1500); - assert.throws(() => client.send("schema", Uint8Array.from([0xc1])), /before the session is connected/, "the grace ran out and the goodbye went out"); - }); -}); - -test("nothing behind the capped frame is served while its answer is delivered", async () => { - const r = await rawRig(new Recording(grant)); - try { - for (let i = 0; i < CAP; i++) r.send(Uint8Array.from([0xc1])); - r.send(encodeWireValue(put(KEY_A))); - await sleep(500); - assert.equal(r.replies.length, CAP); - assert.deepEqual(r.puts, [], "the put behind the cap was not served"); - } finally { r.close(); } -}); diff --git a/packages/cultnet-py/src/cultnet_py/client.py b/packages/cultnet-py/src/cultnet_py/client.py index 35a960e3..28060753 100644 --- a/packages/cultnet-py/src/cultnet_py/client.py +++ b/packages/cultnet-py/src/cultnet_py/client.py @@ -58,7 +58,6 @@ class CultNetRawClient: create_transport: CultNetSchemaTransportFactory | None = None def send(self, message: CultNetMessage | dict[str, Any]) -> None: - """Fire-and-forget: an ack or a return is not admission; read the result frame.""" wire = message.to_wire() if isinstance(message, CultNetMessage) else message with self.open_transport() as transport: transport.send("schema", msgpack.packb(wire, use_bin_type=True)) @@ -320,7 +319,6 @@ class CultNetDatabaseSubscription: self._transport = None def send(self, message: CultNetMessage | dict[str, Any]) -> None: - """Fire-and-forget: an ack or a return is not admission; read the result frame.""" if self._transport is None: raise RuntimeError("CultNet database subscription is not open") wire = message.to_wire() if isinstance(message, CultNetMessage) else message diff --git a/packages/cultnet-py/src/cultnet_py/transport.py b/packages/cultnet-py/src/cultnet_py/transport.py index 801d1313..3ecfa31b 100644 --- a/packages/cultnet-py/src/cultnet_py/transport.py +++ b/packages/cultnet-py/src/cultnet_py/transport.py @@ -1256,7 +1256,6 @@ class CultNetRudpSocketTransportConnection: self._send_packet(self.session.create_connect(_now_ms(), payload)) def send(self, channel_id: str, payload: bytes) -> None: - """Fire-and-forget: an ack or a return is not admission; read the result frame.""" packets = self.session.send_many( channel_id, payload, @@ -1568,7 +1567,6 @@ class TcpFramedTransportConnection: ) def send(self, channel_id: str, payload: bytes) -> None: - """Fire-and-forget: an ack or a return is not admission; read the result frame.""" if channel_id != "schema": raise ValueError(f"tcp_framed transport only supports the schema channel, got {channel_id!r}") before = len(payload) + 4 diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 6a2aa20b..37bf73d8 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -2198,7 +2198,6 @@ impl CultNetRudpSocketTransportConnection { self.send_packet(&packet) } - /// Fire-and-forget: an ack or a return is not admission; read the result frame. pub fn send(&mut self, channel_id: &str, payload: Vec) -> Result<()> { let options = channel_send_options( @@ -2217,8 +2216,6 @@ impl CultNetRudpSocketTransportConnection { /// Sends one frame on a reliable channel whose packets cannot expire and /// returns the transport receipt used to observe its exact acknowledgement. - /// - /// Fire-and-forget: an ack or a return is not admission; read the result frame. pub fn send_reliable( &mut self, channel_id: &str, @@ -2312,9 +2309,6 @@ impl CultNetRudpSocketTransportConnection { self.session.outstanding_reliable_packet_count() } - /// Waits until every reliable write is acknowledged by the transport. - /// - /// Fire-and-forget: an ack or a return is not admission; read the result frame. pub fn flush_reliable(&mut self, timeout: Duration) -> Result<()> { let original_timeout = self.socket.read_timeout()?; let poll_timeout = original_timeout diff --git a/packages/cultnet-rs/src/transport.rs b/packages/cultnet-rs/src/transport.rs index 57070dc3..f2c196d4 100644 --- a/packages/cultnet-rs/src/transport.rs +++ b/packages/cultnet-rs/src/transport.rs @@ -252,7 +252,6 @@ impl TcpFramedTransportConnection where TStream: Write, { - /// Fire-and-forget: an ack or a return is not admission; read the result frame. pub fn send(&mut self, channel_id: &str, payload: &[u8]) -> Result<()> { if channel_id != "schema" { return Err(anyhow!( diff --git a/packages/cultnet-ts/src/idunn-odin-presence-publisher.ts b/packages/cultnet-ts/src/idunn-odin-presence-publisher.ts index 209a2881..3713363d 100644 --- a/packages/cultnet-ts/src/idunn-odin-presence-publisher.ts +++ b/packages/cultnet-ts/src/idunn-odin-presence-publisher.ts @@ -107,36 +107,26 @@ async function publishDocument( const wirePayload = encode(encodeCultNetMessageForWire(message as never, "cultnet.schema.v0")); const packets = session.sendMany("schema", wirePayload, { reliable: true, ordered: true, nowMs: Date.now() }); for (const packet of packets) await sendPacket(socket, endpoint, packet); - // The transport ack is not admission: the put completes on Odin's result frame. A refusal rejects whether - // it arrives before the ack, after it, or inside the result window; a success reply ends the wait early. - let succeeded = false; - const onFrame = (frame: { channelId: string; payload: Uint8Array }) => { - if (frame.channelId !== "schema") return; - const response = parseCultNetMessage(decode(frame.payload)); - if (response.schemaVersion === "cultnet.error.v0") throw new IdunnPresenceRefusedError(response); - succeeded = true; - }; - await receiveUntil(receiver, session, endpoint, () => session.outstandingReliablePacketCount === 0, 2000, onFrame); - if (!succeeded) await receiveUntil(receiver, session, endpoint, () => succeeded, RESULT_WINDOW_MS, onFrame, true); + await receiveUntil( + receiver, + session, + endpoint, + () => session.outstandingReliablePacketCount === 0, + 2000, + (frame) => { + if (frame.channelId !== "schema") return; + const response = parseCultNetMessage(decode(frame.payload)); + if (response.schemaVersion === "cultnet.error.v0") { + throw new Error(`Odin rejected runtime presence: ${response.error}`); + } + }, + ); } finally { receiver.close(); socket.close(); } } -/** How long after the transport ack Odin's refusal may still arrive; a success reply ends the wait early. */ -const RESULT_WINDOW_MS = 500; - -/** Odin refused the presence put: the typed `cultnet.error.v0` it answered with. */ -export class IdunnPresenceRefusedError extends Error { - readonly code: string | undefined; - constructor(readonly response: { error: string; code?: string | null }) { - super(`Odin rejected runtime presence: ${response.error}`); - this.name = "IdunnPresenceRefusedError"; - this.code = response.code ?? undefined; - } -} - type PacketReceiver = { socket: dgram.Socket; next(timeoutMs: number): Promise; @@ -200,7 +190,6 @@ async function receiveUntil( predicate: (packet: CultNetRudpPacket) => boolean, timeoutMs: number, onDelivered?: (frame: { channelId: string; payload: Uint8Array; sequence: number }) => void, - returnOnTimeout = false, ): Promise { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { @@ -215,7 +204,6 @@ async function receiveUntil( } for (const packet of session.dueResends(Date.now())) await sendPacket(receiver.socket, endpoint, packet); } - if (returnOnTimeout) return; throw new Error(`Timed out waiting for CultNet RUDP response after ${timeoutMs} ms.`); } diff --git a/packages/cultnet-ts/src/peer.ts b/packages/cultnet-ts/src/peer.ts index 504973af..0e54ee58 100644 --- a/packages/cultnet-ts/src/peer.ts +++ b/packages/cultnet-ts/src/peer.ts @@ -95,10 +95,6 @@ export class CultNetPeer extends EventEmitter { return this.#transport?.profile; } - /** - * Fire-and-forget: an ack or a return is not admission; read the result frame (a write is refused with - * `cultnet.error.v0`, `write_not_authorized` or `message_invalid`, as a `message` event). - */ send(message: CultNetMessage): void { const wireValue = encodeCultNetMessageForWire(message, this.#wireContract); const payload = encode(wireValue); @@ -110,7 +106,6 @@ export class CultNetPeer extends EventEmitter { this.#stream?.write(encodeFrame(payload)); } - /** Resolves when the transport has acknowledged what was sent. An ack is not admission; read the result frame. */ async flush(timeoutMs?: number): Promise { await this.#transport?.flush?.(timeoutMs); } diff --git a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts b/packages/cultnet-ts/test/idunn-runtime-authority.test.ts index 2e05bbf4..2b45e3b1 100644 --- a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts +++ b/packages/cultnet-ts/test/idunn-runtime-authority.test.ts @@ -276,14 +276,9 @@ async function startOdinPeer(context: TestContext, endpoint: string) { peer.once("error", reject); peer.bind(port, "127.0.0.1", resolve); }); - // The second packet of a split delivery is a timer; the test's end cancels it rather than racing the closed socket. - const timers = new Set(); - const later = (send: () => void) => { timers.add(setTimeout(send, 120)); }; - context.after(() => { for (const timer of timers) clearTimeout(timer); peer.close(); }); + context.after(() => peer.close()); const state = { reject: false, - // How a refusal and the ack of the put reach the publisher: in one packet, or the ack first, or the refusal first. - order: "piggyback" as "piggyback" | "ack-then-refusal" | "refusal-then-ack", received: [] as Array<{ schemaVersion?: string; document: { payload: DocumentPayload } }>, responseAck: undefined as number | undefined, failure: undefined as Error | undefined, @@ -310,15 +305,7 @@ async function startOdinPeer(context: TestContext, endpoint: string) { }); assert.ok(response); state.responseAck = response.ack; - if (state.reject && state.order === "ack-then-refusal") { - peer.send(encodeRudpPacket(serverSession.createAckForReceived(packet.sequence)), remote.port, remote.address); - later(() => peer.send(encodeRudpPacket(response), remote.port, remote.address)); - } else if (state.reject && state.order === "refusal-then-ack") { - peer.send(encodeRudpPacket({ ...response, ack: 0, ackMask: 0 }), remote.port, remote.address); - later(() => peer.send(encodeRudpPacket(serverSession!.createAckForReceived(packet.sequence)), remote.port, remote.address)); - } else { - peer.send(encodeRudpPacket(response), remote.port, remote.address); - } + peer.send(encodeRudpPacket(response), remote.port, remote.address); } catch (error) { state.failure = error as Error; } @@ -345,19 +332,6 @@ test("the Odin publisher publishes the signer's document to the endpoint Expecte assert.equal(peer.failure, undefined, peer.failure?.message); }); -for (const order of ["ack-then-refusal", "refusal-then-ack"] as const) { - test(`the Odin publisher does not complete on the ack of a refused presence (${order})`, async (context) => { - const endpoint = expectedFacts("service").odinEndpoint!; - const peer = await startOdinPeer(context, endpoint); - peer.reject = true; - peer.order = order; - const signer = signerFor(context, "service"); - signer.reportHealth("active"); - const publisher = createIdunnRuntimePresencePublisher({ signer, endpoint }); - await assert.rejects(publisher.publish("denied presence"), /Odin rejected runtime presence: test admission denied/); - }); -} - test("the Odin publisher leaves a capability shortfall to the Rust correlation instead of refusing to publish", async (context) => { const endpoint = expectedFacts("service").odinEndpoint!; const peer = await startOdinPeer(context, endpoint); diff --git a/src/GameCult.Mesh/CultMeshClient.cs b/src/GameCult.Mesh/CultMeshClient.cs index 5e8ef38d..5db03ca5 100644 --- a/src/GameCult.Mesh/CultMeshClient.cs +++ b/src/GameCult.Mesh/CultMeshClient.cs @@ -44,12 +44,6 @@ namespace GameCult.Mesh /// Gets or sets the deadline for one correlated typed operation response. public TimeSpan OperationResponseTimeout { get; set; } = TimeSpan.FromSeconds(10); - - /// - /// Gets or sets how long after a submitted document the provider's refusal may still arrive. The provider - /// sends no success frame for an admitted submission, so this window bounds SubmitDocumentAsync. - /// - public TimeSpan SubmitResultWindow { get; set; } = CultNetWriteCompletion.DefaultResultWindow; } /// @@ -155,7 +149,6 @@ namespace GameCult.Mesh private readonly CultMeshSessionManager _sessions; private readonly TimeSpan _subscriptionResponseTimeout; private readonly TimeSpan _operationResponseTimeout; - private readonly TimeSpan _submitResultWindow; private readonly ConcurrentDictionary _documents = new(StringComparer.Ordinal); private readonly ConcurrentDictionary _collections = new(StringComparer.Ordinal); private bool _disposed; @@ -198,7 +191,6 @@ namespace GameCult.Mesh options.Sessions); _subscriptionResponseTimeout = options.SubscriptionResponseTimeout; _operationResponseTimeout = options.OperationResponseTimeout; - _submitResultWindow = options.SubmitResultWindow; } /// Connects to one stable Verse/provider target using one application protocol. @@ -327,10 +319,7 @@ namespace GameCult.Mesh /// /// Submits one typed document to a provider by stable identity. Submission is uncommitted input; - /// only a provider-authored receipt or resulting state can establish acceptance. The task completes on - /// the provider's result frame: a refusal (write_not_authorized, message_invalid, ...) throws - /// , and a submission not refused within - /// completes. + /// only a provider-authored receipt or resulting state can establish acceptance. /// public async Task SubmitDocumentAsync( CultMeshSessionTarget target, @@ -362,11 +351,7 @@ namespace GameCult.Mesh var session = await ConnectAsync(target, CultMeshProtocols.Documents, cancellationToken) .ConfigureAwait(false); cancellationToken.ThrowIfCancellationRequested(); - await CultNetWriteCompletion.SendAndAwaitResultAsync( - refuse => session.OnCultNet(refuse), - () => session.SendCultNet(message), - _submitResultWindow, - cancellationToken).ConfigureAwait(false); + session.OpenSchemaClient().SendCultNet(message); } /// diff --git a/src/GameCult.Mesh/CultMeshSessions.cs b/src/GameCult.Mesh/CultMeshSessions.cs index 9e06c196..9768dfd5 100644 --- a/src/GameCult.Mesh/CultMeshSessions.cs +++ b/src/GameCult.Mesh/CultMeshSessions.cs @@ -352,7 +352,6 @@ namespace GameCult.Mesh internal long PhysicalGeneration => Interlocked.Read(ref _physicalGeneration); public Observable WatchState() => _states; public ICultNetSchemaClient OpenSchemaClient() => new SessionSchemaClient(this); - /// Fire-and-forget: an ack or a return is not admission; read the result frame. public void SendCultNet(T message) where T : ICultNetSchemaMessage => Channel.SendCultNet(message); public IDisposable OnCultNet(Action callback) where T : ICultNetSchemaMessage { diff --git a/src/GameCult.Networking/CultNetRudpSchemaServer.cs b/src/GameCult.Networking/CultNetRudpSchemaServer.cs index 3378036a..de8d0154 100644 --- a/src/GameCult.Networking/CultNetRudpSchemaServer.cs +++ b/src/GameCult.Networking/CultNetRudpSchemaServer.cs @@ -1,5 +1,4 @@ using System; -using System.Runtime.CompilerServices; using System.Collections.Concurrent; using System.Collections.Generic; using System.Net; @@ -114,24 +113,9 @@ namespace GameCult.Networking private readonly CultNetRudpSocketTransportServer _transport; private readonly ConcurrentDictionary _handlers = new(); private readonly ConcurrentDictionary _schemaPeerAdapters = new(); - private readonly ConditionalWeakTable> _undecodableFramesInARow = new(); private bool _disposed; private ILogger _logger = new NullLogger(); - /// - /// How many consecutive undecodable frames a peer's session may send before it is ended. Fewer are - /// answered with a typed message_invalid and the session is kept; the frame that reaches this - /// number is answered too, then the session ends. Any frame that decodes resets the run. The same - /// number is the cap in every runtime. - /// - public const int MaxConsecutiveRefusedFrames = 8; - - /// - /// How long a session whose capped frame was answered waits for the peer to acknowledge the answer - /// before it ends anyway. - /// - internal const long CappedAnswerGraceMs = 2000; - /// /// Gets or sets the logger used by the server. R-AM: the dispatch backstop in DispatchAsync /// logs every untyped fault through this instead of swallowing it silently. @@ -316,9 +300,6 @@ namespace GameCult.Networking { if (!string.Equals(delivered.Frame.ChannelId, "schema", StringComparison.Ordinal)) return false; - // The capped frame was answered: nothing after it is served while the answer is delivered. - if (delivered.Peer.EndingReason != null) - return false; // A peer's message that cannot be decoded (malformed bytes, an unknown schemaVersion) or // handled ends that peer's session, never the poll: the drain goes on serving every other @@ -336,23 +317,11 @@ namespace GameCult.Networking } catch (Exception ex) { - // The peer sent something no runtime can read: the refusal is the peer's. It is answered - // message_invalid and the session kept, until MaxConsecutiveRefusedFrames in a row end it. - var run = _undecodableFramesInARow.GetOrCreateValue(delivered.Peer); - run.Value++; - _transport.SendSchemaMessage(delivered.Peer, CultNetErrorMessage.ForMessageInvalid("message", "CultNet RUDP schema server could not read the message.")); - if (run.Value < MaxConsecutiveRefusedFrames) - { - Logger.LogError($"CultNet RUDP schema server could not decode a peer's message ({ex.GetType().Name}); it was refused and that peer's session kept."); - return false; - } - // The answer was sent reliably: the session ends once the peer has acknowledged it (a lost - // datagram is resent first), or after the grace. - _transport.EndPeerSessionWhenAnswered(delivered.Peer, CultNetRudpSession.RefusedPacketReason, CappedAnswerGraceMs); - Logger.LogError($"CultNet RUDP schema server could not decode a peer's message ({ex.GetType().Name}); that peer's session ends once its answer is acknowledged."); + // The peer sent something no runtime can read: the refusal is the peer's. + var ended = _transport.EndPeerSession(delivered.Peer, CultNetRudpSession.RefusedPacketReason); + Logger.LogError($"CultNet RUDP schema server could not decode a peer's message ({ex.GetType().Name}); {SessionOutcome(ended)}"); return false; } - _undecodableFramesInARow.GetOrCreateValue(delivered.Peer).Value = 0; if (!_handlers.TryGetValue(message.GetType(), out var handler) || handler == null) { diff --git a/src/GameCult.Networking/CultNetSchemaClient.cs b/src/GameCult.Networking/CultNetSchemaClient.cs index 83865e31..5c6117aa 100644 --- a/src/GameCult.Networking/CultNetSchemaClient.cs +++ b/src/GameCult.Networking/CultNetSchemaClient.cs @@ -26,8 +26,7 @@ namespace GameCult.Networking void Connect(string host, int port); /// - /// Sends one schema-v0 message through the transport. Fire-and-forget: an ack or a return is not admission; - /// read the result frame (). + /// Sends one schema-v0 message through the transport. /// void SendCultNet(T message) where T : ICultNetSchemaMessage; diff --git a/src/GameCult.Networking/CultNetShardWriteForwarding.cs b/src/GameCult.Networking/CultNetShardWriteForwarding.cs index 0fa6c4f7..7038b0a7 100644 --- a/src/GameCult.Networking/CultNetShardWriteForwarding.cs +++ b/src/GameCult.Networking/CultNetShardWriteForwarding.cs @@ -68,12 +68,6 @@ namespace GameCult.Networking /// rudp:// endpoints use the C# RUDP schema client and cultnet:// endpoints use LiteNetLib. /// public Func? CreateClient { get; set; } - - /// - /// Gets or sets how long after a forwarded write the primary's refusal may still arrive; a refusal fails the - /// forward with . Defaults to . - /// - public TimeSpan ResultWindow { get; set; } = CultNetWriteCompletion.DefaultResultWindow; } /// @@ -91,10 +85,7 @@ namespace GameCult.Networking _options = options ?? new CultNetSchemaWriteForwarderOptions(); } - /// - /// Forwards a put and completes on the primary's result frame: a refusal throws - /// , the same failure path as an unreachable primary. - /// + /// public async Task ForwardPutAsync(CultNetShardDescriptor shard, CultNetDocumentPutRawMessage message) { if (message == null) throw new ArgumentNullException(nameof(message)); @@ -104,10 +95,7 @@ namespace GameCult.Networking await SendAsync(endpoint, message).ConfigureAwait(false); } - /// - /// Forwards a delete and completes on the primary's result frame: a refusal throws - /// . - /// + /// public async Task ForwardDeleteAsync(CultNetShardDescriptor shard, CultNetDocumentDeleteMessage message) { if (message == null) throw new ArgumentNullException(nameof(message)); @@ -144,14 +132,7 @@ namespace GameCult.Networking using var client = CreateClient(endpoint); client.Connect(host, port); await WaitForConnectionAsync(client, endpoint).ConfigureAwait(false); - await CultNetWriteCompletion.SendAndAwaitResultAsync( - refuse => - { - client.OnCultNet(refuse); - return null; - }, - () => client.SendCultNet(message), - _options.ResultWindow).ConfigureAwait(false); + client.SendCultNet(message); } private ICultNetSchemaClient CreateClient(string endpoint) diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index da465326..ac2c2952 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -2759,11 +2759,6 @@ namespace GameCult.Networking // under SessionGate by whichever thread's send found the failure; the polling thread reads // it when it retires the peer. internal byte[]? UnsendableReason { get; set; } - - // Set once the peer's capped frame has been answered: the session ends with this reason when the - // peer has acknowledged every answer, or at the deadline. Under SessionGate. - internal byte[]? EndingReason { get; set; } - internal long EndingDeadlineMs { get; set; } } /// @@ -2933,7 +2928,6 @@ namespace GameCult.Networking public bool TryReceiveOnce(out CultNetRudpSocketServerFrame? delivered) { RetireUnsendablePeers(); - EndAnsweredPeers(); if (_deliveredFrames.Count > 0) { delivered = _deliveredFrames.Dequeue(); @@ -3152,7 +3146,6 @@ namespace GameCult.Networking public void PollResends() { RetireUnsendablePeers(); - EndAnsweredPeers(); if (_socket.Poll(0, SelectMode.SelectRead)) return; foreach (var peer in _peers.Values.ToArray()) @@ -3251,44 +3244,6 @@ namespace GameCult.Networking return true; } - // Polling thread only. The application answered a peer's capped frame: the answer is reliable, so a - // lost datagram is resent, and the session ends with `reason` once the peer has acknowledged every - // answer or `graceMs` have passed (EndAnsweredPeers). Frames of the peer still queued are not delivered. - internal void EndPeerSessionWhenAnswered(CultNetRudpSocketServerPeer peer, byte[] reason, long graceMs) - { - if (peer == null) throw new ArgumentNullException(nameof(peer)); - if (reason == null) throw new ArgumentNullException(nameof(reason)); - lock (peer.SessionGate) - { - if (peer.Session.Ended || peer.EndingReason != null) - return; - peer.EndingReason = reason; - peer.EndingDeadlineMs = NowMs() + graceMs; - } - var kept = _deliveredFrames.Where(frame => !ReferenceEquals(frame.Peer, peer)).ToArray(); - _deliveredFrames.Clear(); - foreach (var frame in kept) - _deliveredFrames.Enqueue(frame); - } - - // Polling thread only. - private void EndAnsweredPeers() - { - foreach (var peer in _peers.Values.ToArray()) - { - byte[]? reason; - lock (peer.SessionGate) - { - reason = peer.EndingReason != null - && (peer.Session.OutstandingReliablePacketCount == 0 || NowMs() >= peer.EndingDeadlineMs) - ? peer.EndingReason - : null; - } - if (reason != null) - EndPeerSession(peer, reason); - } - } - // Polling thread only. A peer a new Connect already replaced was reported then, and its key now // names the peer that replaced it. private void RetireUnsendablePeers() diff --git a/src/GameCult.Networking/CultNetWriteCompletion.cs b/src/GameCult.Networking/CultNetWriteCompletion.cs deleted file mode 100644 index 6d5c226d..00000000 --- a/src/GameCult.Networking/CultNetWriteCompletion.cs +++ /dev/null @@ -1,64 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; - -namespace GameCult.Networking -{ - /// - /// A peer refused a write: the typed cultnet.error.v0 it answered with. - /// - public sealed class CultNetWriteRefusedException : Exception - { - /// Creates the exception from the refusal frame. - public CultNetWriteRefusedException(CultNetErrorMessage refusal) - : base(refusal?.Error ?? throw new ArgumentNullException(nameof(refusal))) - { - Code = refusal.Code; - Refusal = refusal; - } - - /// Gets the machine-readable refusal code, e.g. write_not_authorized or message_invalid. - public string? Code { get; } - - /// Gets the refusal frame as received. - public CultNetErrorMessage Refusal { get; } - } - - /// - /// Completes a write on the peer's result frame, never on a send's return or a transport acknowledgement: - /// an ack proves delivery, not admission. - /// - public static class CultNetWriteCompletion - { - /// How long after a send a refusal may still arrive when the caller names no window. - public static readonly TimeSpan DefaultResultWindow = TimeSpan.FromMilliseconds(500); - - /// - /// Sends one write and waits for the peer's answer. A refusal frame rejects with - /// whatever its order against the transport acknowledgement. A peer - /// that admits a write sends no success frame, so the end of the window is the interim bound on a refusal - /// (until peers send one); the error frame carries no message id, so any refusal inside the window is this - /// write's. - /// - /// Registers the refusal handler and returns its registration; called before the send. - /// Sends the write. - /// How long a refusal may take to arrive. - /// Cancels the wait. - public static async Task SendAndAwaitResultAsync( - Func, IDisposable?> subscribe, - Action send, - TimeSpan resultWindow, - CancellationToken cancellationToken = default) - { - if (subscribe == null) throw new ArgumentNullException(nameof(subscribe)); - if (send == null) throw new ArgumentNullException(nameof(send)); - var refused = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - using var registration = subscribe(error => refused.TrySetResult(error)); - send(); - var window = Task.Delay(resultWindow, cancellationToken); - await Task.WhenAny(refused.Task, window).ConfigureAwait(false); - if (refused.Task.IsCompleted) throw new CultNetWriteRefusedException(await refused.Task.ConfigureAwait(false)); - await window.ConfigureAwait(false); - } - } -} diff --git a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs index 9b6e3db7..dcabf6e6 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs @@ -291,51 +291,6 @@ public sealed class CultMeshClientTests submitted.Document.SourceRole.Should().Be("pilot"); } - [TestCase("write_not_authorized", 0)] - [TestCase("message_invalid", 0)] - [TestCase("write_not_authorized", 150)] - public async Task SubmitDocument_ARefusalRejectsWithTheTypedError_WhateverItsOrderAgainstTheSend(string code, int delayMs) - { - var connector = new DocumentConnector - { - RefuseSubmissionsWith = code, - RefusalDelay = TimeSpan.FromMilliseconds(delayMs) - }; - using var mesh = new CultMeshClient(new CultMeshClientOptions - { - RendezvousEndpoints = new[] { "rudp://odin:3076" }, - Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() }, - Sessions = CultMeshTestTrust.LocalSessions, - Connectors = new[] { connector } - }); - var request = new ClientTestDocument { Id = "command:1", Text = "move" }; - - var act = () => mesh.SubmitDocumentAsync(Target, request.Id, request, "pilot-one", "pilot"); - - (await act.Should().ThrowAsync()).Which.Code.Should().Be(code); - } - - [Test] - public async Task SubmitDocument_CompletesOnlyAfterTheResultWindow_NotOnTheSendReturning() - { - var connector = new DocumentConnector(); - using var mesh = new CultMeshClient(new CultMeshClientOptions - { - RendezvousEndpoints = new[] { "rudp://odin:3076" }, - Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() }, - Sessions = CultMeshTestTrust.LocalSessions, - Connectors = new[] { connector }, - SubmitResultWindow = TimeSpan.FromMilliseconds(400) - }); - var request = new ClientTestDocument { Id = "command:1", Text = "move" }; - await mesh.ConnectAsync(Target, CultMeshProtocols.Documents); - var started = DateTime.UtcNow; - - await mesh.SubmitDocumentAsync(Target, request.Id, request, "pilot-one", "pilot"); - - (DateTime.UtcNow - started).Should().BeGreaterThanOrEqualTo(TimeSpan.FromMilliseconds(380)); - } - [Test] public async Task InvokeOperation_CorrelatesProviderReceiptThroughStableIdentitySession() { @@ -541,8 +496,6 @@ public sealed class CultMeshClientTests public bool SuppressFirstSnapshot { get; set; } public bool SuppressFirstSubscriptionSnapshot { get; set; } public bool RejectOperations { get; set; } - public string? RefuseSubmissionsWith { get; set; } - public TimeSpan RefusalDelay { get; set; } public bool FailFirstOperationBeforeResponse { get; set; } public string ResponseSourceRuntimeId { get; set; } = "aetheria-daemon"; public string ConnectorId => "document-test"; @@ -562,8 +515,6 @@ public sealed class CultMeshClientTests rejectOperations: RejectOperations, failOperationBeforeResponse: FailFirstOperationBeforeResponse && Clients.Count == 0, responseSourceRuntimeId: ResponseSourceRuntimeId); - client.RefuseSubmissionsWith = RefuseSubmissionsWith; - client.RefusalDelay = RefusalDelay; Clients.Add(client); return Task.FromResult(client); } @@ -571,9 +522,6 @@ public sealed class CultMeshClientTests private sealed class DocumentClient : ICultNetSchemaClient, ICultNetSchemaClientHealth, ICultMeshVerifiedSchemaClient { - private readonly List> _errors = new(); - public string? RefuseSubmissionsWith { get; set; } - public TimeSpan RefusalDelay { get; set; } private readonly List> _snapshots = new(); private readonly List> _operationResponses = new(); private readonly CultNetDocumentRegistry _documents; @@ -610,15 +558,6 @@ public sealed class CultMeshClientTests if (message is CultNetDocumentPutRawMessage submitted) { Submitted.Add(submitted); - if (RefuseSubmissionsWith != null) - { - var refusal = new CultNetErrorMessage { Error = "not allowed", Code = RefuseSubmissionsWith }; - // A refusal that follows the send's return (and any transport ack) must still reject the submit. - Task.Delay(RefusalDelay).ContinueWith(_ => - { - foreach (var handler in _errors.ToArray()) handler(refusal); - }); - } return; } if (message is CultNetOperationRequestMessage operation) @@ -703,8 +642,6 @@ public sealed class CultMeshClientTests _snapshots.Add(response => callback((T)(object)response)); if (typeof(T) == typeof(CultNetOperationResponseMessage)) _operationResponses.Add(response => callback((T)(object)response)); - if (typeof(T) == typeof(CultNetErrorMessage)) - _errors.Add(error => callback((T)(object)error)); } public void Dispose() { } public void Fail(Exception error) => _failure.TrySetResult(error); diff --git a/tests/GameCult.Networking.Tests/CultNetClientWriteCompletionTests.cs b/tests/GameCult.Networking.Tests/CultNetClientWriteCompletionTests.cs deleted file mode 100644 index 8641a082..00000000 --- a/tests/GameCult.Networking.Tests/CultNetClientWriteCompletionTests.cs +++ /dev/null @@ -1,100 +0,0 @@ -#nullable enable -using System; -using System.Collections.Concurrent; -using System.Net; -using System.Net.Sockets; -using System.Threading; -using System.Threading.Tasks; -using NUnit.Framework; - -namespace GameCult.Networking.Tests -{ - /// - /// Heimdall's invariant: no client treats a reliable-transport ack as write admission. A put completes - /// only on its result frame (write_not_authorized, or the success reply). The C# schema client has no - /// completion call: SendCultNet returns once the put is queued, and the outcome reaches the caller only - /// as the result frame, through OnCultNet. These tests serve it a peer that acknowledges the put and - /// refuses it, the ack before the refusal and after it, and require the refusal to arrive as the outcome. - /// - public class CultNetClientWriteCompletionTests - { - private const uint ConnectionId = 0x10203197; - - // Acknowledges the first reliable schema frame it receives and answers it with write_not_authorized, - // the second packet 150 ms after the first. The refusal carries no acknowledgement of its own. - private static Task RefusingPeer(Socket socket, bool refusalFirst, CancellationToken stop) - { - return Task.Run(() => - { - var session = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId, ResendDelayMs = 25 }); - var buffer = new byte[65535]; - EndPoint from = new IPEndPoint(IPAddress.Any, 0); - var answered = false; - while (!stop.IsCancellationRequested) - { - if (!socket.Poll(10_000, SelectMode.SelectRead)) - continue; - var received = socket.ReceiveFrom(buffer, ref from); - var packet = CultNetRudpPacketCodec.Decode(buffer.AsSpan(0, received).ToArray()); - void Send(CultNetRudpPacket p) => socket.SendTo(CultNetRudpPacketCodec.Encode(p), from); - if (packet.PacketType == CultNetRudpPacketType.Connect) - { - Send(session.AcceptConnect(packet)); - continue; - } - var result = session.Receive(packet, 1); - if (result.Reply != null) - Send(result.Reply); - if (answered || result.Delivered.Count == 0) - continue; - answered = true; - var ack = session.CreateAckForReceived(packet.Sequence); - var refusal = session.Send("schema", CultNetSchemaMessageSerialization.Serialize(new CultNetErrorMessage - { - Error = "write_not_authorized: the catalog refused the document", - Code = "write_not_authorized" - }), new CultNetRudpSendOptions { Reliable = true, Ordered = true }); - refusal.Ack = 0; - refusal.AckMask = 0; - if (refusalFirst) - { - Send(refusal); - Thread.Sleep(150); - Send(ack); - } - else - { - Send(ack); - Thread.Sleep(150); - Send(refusal); - } - } - }); - } - - [TestCase(false, TestName = "AckThenRefusal")] - [TestCase(true, TestName = "RefusalThenAck")] - public async Task ARefusedPutIsReportedRefusedWhateverTheOrderOfItsAck(bool refusalFirst) - { - using var serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp); - serverSocket.Bind(new IPEndPoint(IPAddress.Loopback, 0)); - using var stop = new CancellationTokenSource(); - var peer = RefusingPeer(serverSocket, refusalFirst, stop.Token); - using var client = new RudpCultNetSchemaClient("completion", ConnectionId); - var outcomes = new BlockingCollection(); - client.OnCultNet(outcomes.Add); - client.Connect("127.0.0.1", ((IPEndPoint)serverSocket.LocalEndPoint!).Port); - var deadline = DateTime.UtcNow.AddSeconds(3); - while (!client.Connected && DateTime.UtcNow < deadline) - await Task.Delay(5); - Assert.That(client.Connected, "fixture: the client connected"); - - client.SendCultNet(new CultNetDocumentPutRawMessage { MessageId = "completion" }); - - Assert.That(outcomes.TryTake(out var refusal, 3000), Is.True, "the result frame reached the caller"); - Assert.That(refusal!.Code, Is.EqualTo("write_not_authorized")); - stop.Cancel(); - await peer; - } - } -} diff --git a/tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs b/tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs index e7900e7c..b8de674f 100644 --- a/tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs @@ -14,9 +14,8 @@ namespace GameCult.Networking.Tests { /// /// A schema server's poll serves every peer whatever one peer sends. A message that cannot be - /// decoded is answered message_invalid and its session kept until MaxConsecutiveRefusedFrames in a - /// row end it; one that cannot be handled ends that peer's session. The fault is logged, and the - /// drain goes on to the other peers. + /// decoded or handled is that peer's failure: its session ends, the drop is counted and logged, + /// and the drain goes on to the other peers. /// public class CultNetRudpSchemaServerPeerInputTests { @@ -162,165 +161,30 @@ namespace GameCult.Networking.Tests } [Test] - public async Task APeerWhoseMessageCannotBeDecodedIsAnsweredAndKept() + public async Task APeerWhoseMessageCannotBeDecodedHasItsSessionEnded() { using var scene = await BadThenGood(Malformed()); + var dropped = scene.Server.Stats.PacketsDropped; try { await scene.Server.PollAvailableAsync(16); } catch (MessagePackSerializationException) { Assert.Fail("the poll threw for a peer's input"); } - Assert.That(scene.Ended, Is.Empty); - Assert.That(scene.Server.Peers, Has.Count.EqualTo(2)); - var refusal = ErrorsReceivedBy(scene.Bad).Single(); - Assert.That(refusal.Code, Is.EqualTo("message_invalid")); - Assert.That(refusal.Details!.Field, Is.EqualTo("message")); + // The frame was already acknowledged, so the session cannot be kept: the peer is told, + // the listener forgets it, and the drop is counted. + Assert.That(scene.Ended, Is.EqualTo(new[] { scene.Bad.EndPoint })); + Assert.That(scene.Server.Peers.Select(peer => peer.RemoteEndPoint), Is.EqualTo(new[] { scene.Good.EndPoint })); + Assert.That(scene.Server.Stats.PacketsDropped, Is.EqualTo(dropped + 1)); + var goodbye = Drain(scene.Bad.Socket).Single(packet => packet.PacketType == CultNetRudpPacketType.Disconnect); + Assert.That(scene.Bad.Session.Receive(goodbye, 1).Disconnected, Is.True); Assert.That(Drain(scene.Good.Socket).Select(packet => packet.PacketType), Has.None.EqualTo(CultNetRudpPacketType.Disconnect)); Assert.That(scene.Logger.Errors, Has.Count.EqualTo(1)); } - private static List ErrorsReceivedBy(Client client) - { - var errors = new List(); - foreach (var packet in Drain(client.Socket)) - { - var received = client.Session.Receive(packet, 1); - foreach (var frame in received.Delivered) - if (CultNetSchemaMessageSerialization.Deserialize(frame.Payload) is CultNetErrorMessage error) - errors.Add(error); - } - return errors; - } - - // The bad peer acknowledges what it received and the server polls on. - private static async Task Acknowledge(Scene scene) - { - scene.Bad.Socket.SendTo(CultNetRudpPacketCodec.Encode(scene.Bad.Session.CreateAck()), scene.Server.LocalEndPoint); - await scene.Server.PollAvailableAsync(16); - } - - // The bad peer has sent seven undecodable frames (BadThenGood sent the first), each answered. - private static async Task AtTheCap() - { - var scene = await BadThenGood(Malformed()); - await scene.Server.PollAvailableAsync(16); - ErrorsReceivedBy(scene.Bad); - for (var i = 0; i < 6; i++) - { - SendSchema(scene.Server, scene.Bad, Malformed()); - await scene.Server.PollAvailableAsync(16); - ErrorsReceivedBy(scene.Bad); - } - Assert.That(scene.Ended, Is.Empty); - return scene; - } - - // The capped frame's answer is lost on the wire: the server resends it, and the goodbye follows only - // once the peer has acknowledged it. - [Test] - public async Task ALostCappedAnswerIsResentAndTheGoodbyeWaitsForItsAcknowledgement() - { - using var scene = await AtTheCap(); - SendSchema(scene.Server, scene.Bad, Malformed()); - await scene.Server.PollAvailableAsync(16); - - var lost = Drain(scene.Bad.Socket); - Assert.That(lost.Where(p => p.PacketType == CultNetRudpPacketType.Data), Is.Not.Empty, "fixture: the answer was sent and is lost here"); - Assert.That(lost.Select(p => p.PacketType), Has.None.EqualTo(CultNetRudpPacketType.Disconnect), "no goodbye before the answer is acknowledged"); - Assert.That(scene.Ended, Is.Empty); - - var resent = new List(); - for (var attempt = 0; attempt < 40 && resent.Count == 0; attempt++) - { - await Task.Delay(50); - scene.Server.PollResends(); - resent = Drain(scene.Bad.Socket); - } - Assert.That(resent.Select(p => p.PacketType), Has.None.EqualTo(CultNetRudpPacketType.Disconnect), "still no goodbye"); - var answers = new List(); - foreach (var packet in resent) - foreach (var frame in scene.Bad.Session.Receive(packet, 1).Delivered) - if (CultNetSchemaMessageSerialization.Deserialize(frame.Payload) is CultNetErrorMessage error) - answers.Add(error); - Assert.That(answers.Select(e => e.Code), Is.EqualTo(new[] { "message_invalid" }), "the lost answer arrived by resend"); - Assert.That(scene.Ended, Is.Empty); - - await Acknowledge(scene); - Assert.That(scene.Ended, Is.EqualTo(new[] { scene.Bad.EndPoint }), "acknowledged: the session ends"); - Assert.That(Drain(scene.Bad.Socket).Select(p => p.PacketType), Has.Some.EqualTo(CultNetRudpPacketType.Disconnect)); - } - - // A peer that never acknowledges the answer is not kept for ever: the session ends after the grace. - [Test] - public async Task ACappedSessionWhoseAnswerIsNeverAcknowledgedEndsAfterTheGrace() - { - using var scene = await AtTheCap(); - SendSchema(scene.Server, scene.Bad, Malformed()); - await scene.Server.PollAvailableAsync(16); - Drain(scene.Bad.Socket); - - await Task.Delay(1500); // the grace is two seconds - scene.Server.PollResends(); - Assert.That(scene.Ended, Is.Empty, "inside the grace"); - await Task.Delay(700); - scene.Server.PollResends(); - Assert.That(scene.Ended, Is.EqualTo(new[] { scene.Bad.EndPoint }), "the grace ran out"); - } - - // Nothing behind the capped frame is served while its answer is delivered. - [Test] - public async Task NothingBehindTheCappedFrameIsServed() - { - using var scene = await AtTheCap(); - SendSchema(scene.Server, scene.Bad, Malformed()); - SendSchema(scene.Server, scene.Bad, CultNetSchemaMessageSerialization.Serialize( - new CultNetShardCatalogRequestMessage { MessageId = "behind-the-cap" })); - await scene.Server.PollAvailableAsync(16); - await scene.Server.PollAvailableAsync(16); - - Assert.That(scene.Served.Select(s => s.MessageId), Does.Not.Contain("behind-the-cap")); - } - - // The numbers are the shared cross-runtime cap, spelled out so changing the constant fails here. - [Test] - public async Task EightConsecutiveUndecodableFramesEndTheSessionAfterTheEighthIsAnswered() - { - Assert.That(RudpCultNetSchemaServer.MaxConsecutiveRefusedFrames, Is.EqualTo(8)); - using var scene = await BadThenGood(Malformed()); - await scene.Server.PollAvailableAsync(16); - ErrorsReceivedBy(scene.Bad); - async Task Refuse(int count) - { - for (var i = 0; i < count; i++) - { - SendSchema(scene.Server, scene.Bad, Malformed()); - await scene.Server.PollAvailableAsync(16); - } - } - - await Refuse(6); // with the first, seven in a row - Assert.That(ErrorsReceivedBy(scene.Bad), Has.Count.EqualTo(6)); - Assert.That(scene.Ended, Is.Empty); - SendSchema(scene.Server, scene.Bad, CultNetSchemaMessageSerialization.Serialize( - new CultNetShardCatalogRequestMessage { MessageId = "resets-the-run" })); - await scene.Server.PollAvailableAsync(16); - Assert.That(scene.Served.Select(s => s.MessageId), Does.Contain("resets-the-run")); - await Refuse(7); - Assert.That(ErrorsReceivedBy(scene.Bad), Has.Count.EqualTo(7)); - Assert.That(scene.Ended, Is.Empty, "a decoded frame reset the run"); - - await Refuse(1); - Assert.That(ErrorsReceivedBy(scene.Bad).Select(e => e.Code), Is.EqualTo(new[] { "message_invalid" }), "the eighth is answered too"); - Assert.That(scene.Ended, Is.Empty, "the session ends once the answer is acknowledged"); - await Acknowledge(scene); - Assert.That(scene.Ended, Is.EqualTo(new[] { scene.Bad.EndPoint })); - Assert.That(scene.Server.Peers.Select(peer => peer.RemoteEndPoint), Is.EqualTo(new[] { scene.Good.EndPoint })); - } - // The peer's malformed message and a well-formed one after it arrive in one datagram's - // delivery: the second is held for order, and both are released together. The session is - // kept through the first, so the second is handled. + // delivery: the second is held for order, and both are released together. The session ends + // on the first, so the second belongs to no session and is never handled. [Test] - public async Task AFrameQueuedBehindAnUndecodableOneIsStillHandled() + public async Task ARefusedPeersFramesStillQueuedAreNotHandled() { var logger = new CapturingLogger(); using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions @@ -341,8 +205,8 @@ namespace GameCult.Networking.Tests await server.PollAvailableAsync(16); - Assert.That(served, Is.EqualTo(new[] { "after-the-refused-one" })); - Assert.That(server.Peers, Has.Count.EqualTo(1)); + Assert.That(served, Is.Empty); + Assert.That(server.Peers, Is.Empty); Assert.That(logger.Errors, Has.Count.EqualTo(1)); bad.Socket.Dispose(); } diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 2a75b5c2..5a1f570f 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -5213,52 +5213,6 @@ namespace GameCult.Networking.Tests Assert.That(sent.ShardEpoch, Is.EqualTo(3)); } - [TestCase("write_not_authorized", 0)] - [TestCase("message_invalid", 0)] - [TestCase("write_not_authorized", 150)] - public void CultNetSchemaWriteForwarder_RejectsWithTheTypedRefusal_WhateverItsOrderAgainstTheSend(string code, int delayMs) - { - var fake = new CapturingSchemaClient(_ => null); - var forwarder = new CultNetSchemaWriteForwarder(new CultNetSchemaWriteForwarderOptions - { - CreateClient = () => fake, - ResultWindow = TimeSpan.FromSeconds(2) - }); - var shard = new CultNetShardDescriptor( - "notes", "primary", 3, isPrimary: false, - primaryEndpoints: ["cultnet://primary.example.test:4075"]); - fake.OnSend = () => - { - var refusal = new CultNetErrorMessage { Error = "refused", Code = code }; - if (delayMs == 0) fake.Deliver(refusal); - else Task.Delay(delayMs).ContinueWith(_ => fake.Deliver(refusal)); - }; - - var put = Assert.ThrowsAsync(() => forwarder.ForwardPutAsync(shard, new CultNetDocumentPutRawMessage())); - Assert.That(put!.Code, Is.EqualTo(code)); - var delete = Assert.ThrowsAsync(() => forwarder.ForwardDeleteAsync(shard, new CultNetDocumentDeleteMessage { SchemaId = "note.v0", RecordKey = "note:1" })); - Assert.That(delete!.Code, Is.EqualTo(code)); - } - - [Test] - public async Task CultNetSchemaWriteForwarder_DoesNotCompleteOnTheSendReturning() - { - var fake = new CapturingSchemaClient(_ => null); - var forwarder = new CultNetSchemaWriteForwarder(new CultNetSchemaWriteForwarderOptions - { - CreateClient = () => fake, - ResultWindow = TimeSpan.FromMilliseconds(400) - }); - var shard = new CultNetShardDescriptor( - "notes", "primary", 3, isPrimary: false, - primaryEndpoints: ["cultnet://primary.example.test:4075"]); - var started = DateTime.UtcNow; - - await forwarder.ForwardDeleteAsync(shard, new CultNetDocumentDeleteMessage { SchemaId = "note.v0", RecordKey = "note:1" }); - - Assert.That(DateTime.UtcNow - started, Is.GreaterThanOrEqualTo(TimeSpan.FromMilliseconds(380))); - } - [Test] public void CultMeshVerseCatalog_Finds_CompatibleTransferTargets() { @@ -7504,14 +7458,6 @@ namespace GameCult.Networking.Tests _respond = respond; } - public Action? OnSend { get; set; } - - public void Deliver(ICultNetSchemaMessage message) - { - if (!_handlers.TryGetValue(message.GetType(), out var handlers)) return; - foreach (var handler in handlers.ToArray()) handler.DynamicInvoke(message); - } - public bool Connected { get; private set; } public string? ConnectedHost { get; private set; } public int ConnectedPort { get; private set; } @@ -7528,7 +7474,6 @@ namespace GameCult.Networking.Tests where T : ICultNetSchemaMessage { SentMessages.Add(message); - OnSend?.Invoke(); var response = _respond(message); if (response == null) {