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 10cc2a58..7d052785 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,6 +808,7 @@ 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? @@ -3167,13 +3168,16 @@ 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 69583c1c..ab5267a7 100644 --- a/packages/cultmesh-py/src/cultmesh_py/server.py +++ b/packages/cultmesh-py/src/cultmesh_py/server.py @@ -59,6 +59,16 @@ 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 @@ -90,6 +100,10 @@ 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 @@ -346,6 +360,7 @@ 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: @@ -394,18 +409,31 @@ 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): - 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"), - ) + 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.")] for response in responses: if unsendable is None: unsendable = self._send_rudp_schema_frame( @@ -415,6 +443,11 @@ 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 @@ -445,11 +478,26 @@ 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 new file mode 100644 index 00000000..0f9d0c70 --- /dev/null +++ b/packages/cultmesh-py/tests/test_client_write_completion.py @@ -0,0 +1,52 @@ +"""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 b49b6ed9..d96c4ab2 100644 --- a/packages/cultmesh-py/tests/test_cultmesh.py +++ b/packages/cultmesh-py/tests/test_cultmesh.py @@ -24,7 +24,9 @@ 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, @@ -4889,6 +4891,140 @@ 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 d7c2c8fd..33cb4858 100644 --- a/packages/cultmesh-rs/src/lib.rs +++ b/packages/cultmesh-rs/src/lib.rs @@ -82,7 +82,10 @@ 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, @@ -149,6 +152,7 @@ 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, @@ -882,77 +886,30 @@ fn unix_millis() -> u128 { /// Publish one raw CultNet schema message to a RUDP catalog. /// -/// 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. +/// 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. pub fn publish_cultnet_message_to_rudp_catalog( message: &cultnet_rs::CultNetMessage, options: CultMeshRudpDocumentPublishOptions, ) -> Result<()> { - 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_cultnet_messages_to_rudp_catalog(std::slice::from_ref(message), options) } -/// Publish a batch of raw CultNet schema messages to a RUDP catalog. +/// Publish a batch of raw CultNet schema messages to a RUDP catalog on one +/// connection. /// -/// 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. +/// 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. pub fn publish_cultnet_messages_to_rudp_catalog( messages: &[cultnet_rs::CultNetMessage], options: CultMeshRudpDocumentPublishOptions, @@ -969,11 +926,62 @@ 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)?; - client.send("schema", payload)?; + 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.flush_reliable(options.flush_timeout) } fn connect_rudp_client( @@ -1451,6 +1459,141 @@ 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. @@ -1901,4 +2044,63 @@ 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 817e9b5c..647589e0 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -19,6 +19,16 @@ 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 @@ -196,7 +206,10 @@ 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. @@ -241,12 +254,18 @@ 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, @@ -287,7 +306,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; 9] = [ +pub(crate) const REFUSAL_TEXTS: [&str; 11] = [ "the catalog refused the document", "the document can never be served", "the snapshot source failed", @@ -297,6 +316,8 @@ pub(crate) const REFUSAL_TEXTS: [&str; 9] = [ "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 { @@ -312,6 +333,8 @@ impl CultMeshRudpRejectionReason { Self::PayloadBudgetFull => 6, Self::ResponseQueueFailed(_) => 7, Self::ResponseSendFailed(_) => 8, + Self::DeleteNotSupported => 9, + Self::MessageUnreadable(_) => 10, }] } } @@ -331,25 +354,46 @@ 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 concerned; `None` for a snapshot request. A sink refusal names it to the peer. + /// 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. pub record_key: Option, pub reason: CultMeshRudpRejectionReason, } impl CultMeshRudpApplicationRejection { - /// 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. + /// 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. fn refusal_message(&self) -> CultNetMessage { let text = self.reason.refusal_text(); match (&self.reason, &self.record_key) { - (CultMeshRudpRejectionReason::SinkRefused(_), Some(record_key)) => { - write_refused_message(record_key, text) + ( + CultMeshRudpRejectionReason::SinkRefused(_) + | CultMeshRudpRejectionReason::DeleteNotSupported, + Some(record_key), + ) => write_refused_message(record_key, text), + (CultMeshRudpRejectionReason::MessageUnreadable(_), _) => { + message_invalid_message("message", text) } _ => CultNetMessage::Error { error: text.into(), @@ -382,6 +426,11 @@ 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. @@ -556,7 +605,18 @@ where }; let mut unsendable = match result.reply { - Some(reply) => self.send_packet(key.remote_addr, &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)? + } None => None, }; @@ -565,24 +625,125 @@ where return Ok(CultMeshRudpPollOutcome::Handled); } + let mut kept_rejection = None; for frame in result.delivered { if frame.channel_id != "schema" { continue; } - let message = match decode_cultnet_message_from_slice( + // 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( &frame.payload, CultNetWireContract::CultNetSchemaV0, - ) { - Ok(message) => message, - Err(_) => continue, + ); + // 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:#}")), + }), }; - if let Some(rejection) = self.deliver_application_message( - key, - frame.sequence, - message, - now_unix, - now_monotonic, - )? { + 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; + } // 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 @@ -597,7 +758,11 @@ where } } - if packet.reliable && unsendable.is_none() { + 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 { let ack = self .sessions .get_mut(&key) @@ -609,7 +774,11 @@ where if let Some(error) = unsendable { self.end_session(key, SessionEnd::Unsendable(error))?; } - Ok(CultMeshRudpPollOutcome::Handled) + self.end_answered_sessions()?; + Ok(match kept_rejection { + Some(rejection) => CultMeshRudpPollOutcome::ApplicationRejected(rejection), + None => CultMeshRudpPollOutcome::Handled, + }) } /// Expire idle sessions and resend reliable packets whose deadline passed. @@ -623,10 +792,16 @@ 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 packet in entry.session.due_resends(now) { + 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; + } resends.push((*key, packet)); } } @@ -645,6 +820,26 @@ 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( @@ -694,6 +889,9 @@ 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 @@ -898,6 +1096,19 @@ 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) @@ -920,7 +1131,7 @@ where }; let (refusal, mut unsendable) = match end { SessionEnd::Refused => (Vec::new(), None), - SessionEnd::Rejected(refusal) => (self.refusal_packets(&mut entry.session, &refusal), None), + SessionEnd::Rejected(refusal) => (self.refusal_packets(&mut entry.session, &refusal, false), None), SessionEnd::Unsendable(error) => (Vec::new(), Some(error)), }; entry.session.reset_peer_state(); @@ -943,11 +1154,12 @@ where Ok(()) } - /// The refusal a rejected peer is sent, as unreliable, unordered packets. + /// The refusal a rejected peer is sent: unreliable and unordered, or (`reliable`) queued like any response. fn refusal_packets( &self, session: &mut CultNetRudpSession, refusal: &CultNetMessage, + reliable: bool, ) -> Vec { encode_cultnet_message_to_vec(refusal, CultNetWireContract::CultNetSchemaV0) .and_then(|payload| { @@ -955,8 +1167,8 @@ where "schema", payload, CultNetRudpSendOptions { - reliable: false, - ordered: false, + reliable, + ordered: reliable, sequenced: false, now_ms: self.clock.now_monotonic_millis(), reliable_expire_after_ms: None, @@ -1129,6 +1341,9 @@ 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() } @@ -1142,4 +1357,14 @@ 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 8d0c864c..0e3cba66 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -343,14 +343,17 @@ fn refusal_seen_by( client: &mut CultNetRudpSocketTransportConnection, receipt: &CultNetRudpReliableSendReceipt, ) -> Result { - let (error, code, details) = typed_refusal_seen_by(client, receipt)?; + let (error, code, details) = typed_refusal_seen_by(client, receipt, None)?; 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 { @@ -381,6 +384,9 @@ 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; @@ -681,7 +687,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)?, + typed_refusal_seen_by(&mut publisher, &publish_receipt, None)?, ( "write_not_authorized: the catalog refused the document".to_string(), Some(CultNetErrorCode::WriteNotAuthorized), @@ -751,6 +757,173 @@ 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); @@ -1456,6 +1629,11 @@ 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, @@ -1743,3 +1921,138 @@ 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 22d99904..d06e753b 100644 --- a/packages/cultmesh-ts/CHANGELOG.md +++ b/packages/cultmesh-ts/CHANGELOG.md @@ -16,6 +16,7 @@ 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 e8d526cc..f065dd83 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -4121,6 +4121,16 @@ 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; @@ -4173,7 +4183,29 @@ 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; @@ -5537,6 +5569,11 @@ 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(); @@ -5591,6 +5628,8 @@ export class CultMesh { connectPayload, closed: false, unsendable: false, + unreadableInARow: 0, + refusedDeletesInARow: 0, session: new CultNetRudpSession({ connectionId, initialSequence: options.initialSequence, @@ -5655,7 +5694,8 @@ export class CultMesh { record: SessionRecord, payload: Uint8Array, ): Promise { - if (record.closed) return; + // Nothing after the capped frame is served while its answer is delivered. + if (record.closed || record.endingDeadlineMs !== undefined) return; let decoded: unknown; let message: CultNetMessage; try { @@ -5665,12 +5705,21 @@ 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); - 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.")); + 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); 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) { @@ -5744,7 +5793,12 @@ export class CultMesh { } return; case "cultnet.document_delete.v0": { - if (await writeRefused(record, "delete", message.schemaId, message.recordKey, null, null)) return; + 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 (!options.onDocumentDeleteRaw) { sendSchemaMessage(record, cultNetTextError("CultMesh RUDP document server has no delete handler.")); return; @@ -5895,6 +5949,17 @@ 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; @@ -6028,6 +6093,11 @@ 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); } @@ -6109,11 +6179,23 @@ 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, - options.messageId ?? `${runtimeId}:${binding.definition.type}:${recordKey}`, + messageId, recordKey, value, { @@ -6124,6 +6206,13 @@ 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(); } @@ -6142,6 +6231,7 @@ 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 new file mode 100644 index 00000000..8dd122ed --- /dev/null +++ b/packages/cultmesh-ts/test/client-write-completion.test.ts @@ -0,0 +1,117 @@ +// 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 46fc5268..3c35171e 100644 --- a/packages/cultmesh-ts/test/remote-write-authority.test.ts +++ b/packages/cultmesh-ts/test/remote-write-authority.test.ts @@ -1,11 +1,13 @@ // 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, @@ -16,6 +18,7 @@ import { } from "@gamecult/cultnet-ts"; import { CultMesh, + CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES, CultMeshGrantAllWrites, type CultMeshRemoteWriteAuthority, type CultMeshRemoteWriteDecision, @@ -505,3 +508,178 @@ 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 28060753..35a960e3 100644 --- a/packages/cultnet-py/src/cultnet_py/client.py +++ b/packages/cultnet-py/src/cultnet_py/client.py @@ -58,6 +58,7 @@ 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)) @@ -319,6 +320,7 @@ 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 3ecfa31b..801d1313 100644 --- a/packages/cultnet-py/src/cultnet_py/transport.py +++ b/packages/cultnet-py/src/cultnet_py/transport.py @@ -1256,6 +1256,7 @@ 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, @@ -1567,6 +1568,7 @@ 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 37bf73d8..6a2aa20b 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -2198,6 +2198,7 @@ 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( @@ -2216,6 +2217,8 @@ 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, @@ -2309,6 +2312,9 @@ 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 f2c196d4..57070dc3 100644 --- a/packages/cultnet-rs/src/transport.rs +++ b/packages/cultnet-rs/src/transport.rs @@ -252,6 +252,7 @@ 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 3713363d..209a2881 100644 --- a/packages/cultnet-ts/src/idunn-odin-presence-publisher.ts +++ b/packages/cultnet-ts/src/idunn-odin-presence-publisher.ts @@ -107,26 +107,36 @@ 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); - 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}`); - } - }, - ); + // 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); } 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; @@ -190,6 +200,7 @@ 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) { @@ -204,6 +215,7 @@ 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 0e54ee58..504973af 100644 --- a/packages/cultnet-ts/src/peer.ts +++ b/packages/cultnet-ts/src/peer.ts @@ -95,6 +95,10 @@ 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); @@ -106,6 +110,7 @@ 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 2b45e3b1..2e05bbf4 100644 --- a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts +++ b/packages/cultnet-ts/test/idunn-runtime-authority.test.ts @@ -276,9 +276,14 @@ async function startOdinPeer(context: TestContext, endpoint: string) { peer.once("error", reject); peer.bind(port, "127.0.0.1", resolve); }); - context.after(() => peer.close()); + // 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(); }); 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, @@ -305,7 +310,15 @@ async function startOdinPeer(context: TestContext, endpoint: string) { }); assert.ok(response); state.responseAck = response.ack; - peer.send(encodeRudpPacket(response), remote.port, remote.address); + 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); + } } catch (error) { state.failure = error as Error; } @@ -332,6 +345,19 @@ 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 5db03ca5..5e8ef38d 100644 --- a/src/GameCult.Mesh/CultMeshClient.cs +++ b/src/GameCult.Mesh/CultMeshClient.cs @@ -44,6 +44,12 @@ 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; } /// @@ -149,6 +155,7 @@ 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; @@ -191,6 +198,7 @@ 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. @@ -319,7 +327,10 @@ 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. + /// 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. /// public async Task SubmitDocumentAsync( CultMeshSessionTarget target, @@ -351,7 +362,11 @@ namespace GameCult.Mesh var session = await ConnectAsync(target, CultMeshProtocols.Documents, cancellationToken) .ConfigureAwait(false); cancellationToken.ThrowIfCancellationRequested(); - session.OpenSchemaClient().SendCultNet(message); + await CultNetWriteCompletion.SendAndAwaitResultAsync( + refuse => session.OnCultNet(refuse), + () => session.SendCultNet(message), + _submitResultWindow, + cancellationToken).ConfigureAwait(false); } /// diff --git a/src/GameCult.Mesh/CultMeshSessions.cs b/src/GameCult.Mesh/CultMeshSessions.cs index 9768dfd5..9e06c196 100644 --- a/src/GameCult.Mesh/CultMeshSessions.cs +++ b/src/GameCult.Mesh/CultMeshSessions.cs @@ -352,6 +352,7 @@ 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 de8d0154..3378036a 100644 --- a/src/GameCult.Networking/CultNetRudpSchemaServer.cs +++ b/src/GameCult.Networking/CultNetRudpSchemaServer.cs @@ -1,4 +1,5 @@ using System; +using System.Runtime.CompilerServices; using System.Collections.Concurrent; using System.Collections.Generic; using System.Net; @@ -113,9 +114,24 @@ 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. @@ -300,6 +316,9 @@ 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 @@ -317,11 +336,23 @@ namespace GameCult.Networking } catch (Exception ex) { - // 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)}"); + // 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."); 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 5c6117aa..83865e31 100644 --- a/src/GameCult.Networking/CultNetSchemaClient.cs +++ b/src/GameCult.Networking/CultNetSchemaClient.cs @@ -26,7 +26,8 @@ namespace GameCult.Networking void Connect(string host, int port); /// - /// Sends one schema-v0 message through the transport. + /// Sends one schema-v0 message through the transport. Fire-and-forget: an ack or a return is not admission; + /// read the result frame (). /// void SendCultNet(T message) where T : ICultNetSchemaMessage; diff --git a/src/GameCult.Networking/CultNetShardWriteForwarding.cs b/src/GameCult.Networking/CultNetShardWriteForwarding.cs index 7038b0a7..0fa6c4f7 100644 --- a/src/GameCult.Networking/CultNetShardWriteForwarding.cs +++ b/src/GameCult.Networking/CultNetShardWriteForwarding.cs @@ -68,6 +68,12 @@ 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; } /// @@ -85,7 +91,10 @@ 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)); @@ -95,7 +104,10 @@ 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)); @@ -132,7 +144,14 @@ namespace GameCult.Networking using var client = CreateClient(endpoint); client.Connect(host, port); await WaitForConnectionAsync(client, endpoint).ConfigureAwait(false); - client.SendCultNet(message); + await CultNetWriteCompletion.SendAndAwaitResultAsync( + refuse => + { + client.OnCultNet(refuse); + return null; + }, + () => client.SendCultNet(message), + _options.ResultWindow).ConfigureAwait(false); } private ICultNetSchemaClient CreateClient(string endpoint) diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index ac2c2952..da465326 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -2759,6 +2759,11 @@ 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; } } /// @@ -2928,6 +2933,7 @@ namespace GameCult.Networking public bool TryReceiveOnce(out CultNetRudpSocketServerFrame? delivered) { RetireUnsendablePeers(); + EndAnsweredPeers(); if (_deliveredFrames.Count > 0) { delivered = _deliveredFrames.Dequeue(); @@ -3146,6 +3152,7 @@ namespace GameCult.Networking public void PollResends() { RetireUnsendablePeers(); + EndAnsweredPeers(); if (_socket.Poll(0, SelectMode.SelectRead)) return; foreach (var peer in _peers.Values.ToArray()) @@ -3244,6 +3251,44 @@ 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 new file mode 100644 index 00000000..6d5c226d --- /dev/null +++ b/src/GameCult.Networking/CultNetWriteCompletion.cs @@ -0,0 +1,64 @@ +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 dcabf6e6..9b6e3db7 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs @@ -291,6 +291,51 @@ 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() { @@ -496,6 +541,8 @@ 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"; @@ -515,6 +562,8 @@ 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); } @@ -522,6 +571,9 @@ 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; @@ -558,6 +610,15 @@ 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) @@ -642,6 +703,8 @@ 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 new file mode 100644 index 00000000..8641a082 --- /dev/null +++ b/tests/GameCult.Networking.Tests/CultNetClientWriteCompletionTests.cs @@ -0,0 +1,100 @@ +#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 b8de674f..e7900e7c 100644 --- a/tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs @@ -14,8 +14,9 @@ namespace GameCult.Networking.Tests { /// /// A schema server's poll serves every peer whatever one peer sends. A message that cannot be - /// 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. + /// 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. /// public class CultNetRudpSchemaServerPeerInputTests { @@ -161,30 +162,165 @@ namespace GameCult.Networking.Tests } [Test] - public async Task APeerWhoseMessageCannotBeDecodedHasItsSessionEnded() + public async Task APeerWhoseMessageCannotBeDecodedIsAnsweredAndKept() { 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"); } - // 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(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")); Assert.That(Drain(scene.Good.Socket).Select(packet => packet.PacketType), Has.None.EqualTo(CultNetRudpPacketType.Disconnect)); Assert.That(scene.Logger.Errors, Has.Count.EqualTo(1)); } - // 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 ends - // on the first, so the second belongs to no session and is never handled. + 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 ARefusedPeersFramesStillQueuedAreNotHandled() + 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. + [Test] + public async Task AFrameQueuedBehindAnUndecodableOneIsStillHandled() { var logger = new CapturingLogger(); using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions @@ -205,8 +341,8 @@ namespace GameCult.Networking.Tests await server.PollAvailableAsync(16); - Assert.That(served, Is.Empty); - Assert.That(server.Peers, Is.Empty); + Assert.That(served, Is.EqualTo(new[] { "after-the-refused-one" })); + Assert.That(server.Peers, Has.Count.EqualTo(1)); 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 5a1f570f..2a75b5c2 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -5213,6 +5213,52 @@ 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() { @@ -7458,6 +7504,14 @@ 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; } @@ -7474,6 +7528,7 @@ namespace GameCult.Networking.Tests where T : ICultNetSchemaMessage { SentMessages.Add(message); + OnSend?.Invoke(); var response = _respond(message); if (response == null) {