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

merge main

Browse source
This commit is contained in:
Libby 2026-11-08 11:33:16 +00:00
commit dc7eec7e6e
4 changed files with 310 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

52
packages/cultmesh-py/tests/test_client_write_completion.py Normal file
View file

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

134
packages/cultmesh-rs/src/lib.rs
View file

@ -1451,6 +1451,126 @@ 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.
fn acknowledging_refusing_catalog(
refusal_first: bool,
) -> Result<(
SocketAddr,
std::sync::Arc<std::sync::atomic::AtomicBool>,
thread::JoinHandle<Result<()>>,
)> {
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<CultNetRudpSession> = 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(
&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 refusal_first {
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 +2021,18 @@ 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(true)?;
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(())
}
}

25
packages/cultnet-ts/test/idunn-runtime-authority.test.ts
View file

@ -279,6 +279,8 @@ async function startOdinPeer(context: TestContext, endpoint: string) {
context.after(() => peer.close());
const state = {
reject: false,
// How a refusal and the ack of the put reach the publisher: in one packet, or the ack first, or the refusal first.
order: "piggyback" as "piggyback" | "ack-then-refusal" | "refusal-then-ack",
received: [] as Array<{ schemaVersion?: string; document: { payload: DocumentPayload } }>,
responseAck: undefined as number | undefined,
failure: undefined as Error | undefined,
@ -305,7 +307,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);
setTimeout(() => peer.send(encodeRudpPacket(response), remote.port, remote.address), 120);
} else if (state.reject && state.order === "refusal-then-ack") {
peer.send(encodeRudpPacket({ ...response, ack: 0, ackMask: 0 }), remote.port, remote.address);
setTimeout(() => peer.send(encodeRudpPacket(serverSession!.createAckForReceived(packet.sequence)), remote.port, remote.address), 120);
} else {
peer.send(encodeRudpPacket(response), remote.port, remote.address);
}
} catch (error) {
state.failure = error as Error;
}
@ -332,6 +342,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 ["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);

100
tests/GameCult.Networking.Tests/CultNetClientWriteCompletionTests.cs Normal file
View file

@ -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
{
/// <summary>
/// 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.
/// </summary>
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<CultNetErrorMessage>();
client.OnCultNet<CultNetErrorMessage>(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;
}
}
}