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

Merge origin/main into hands/ack-cutD: Cut 3 flow window, RUDP client bind

Browse source 
Import lists in cultnet-ts index.ts and operation-service.ts and the
cultnet-py test block conflicted; both sides are kept.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 15:49:04 +02:00
commit 2e94dc3aa3
25 changed files with 2260 additions and 87 deletions
Download patch file Download diff file Expand all files Collapse all files

644
docs/cultnet-error-contract-cut.md Normal file
View file

@ -0,0 +1,644 @@
# CultNet error contract: cut map
Status: cut map, not started. Imagination pass 1, 2026-09-30, against CultLib `main` at `8fc74c70`.
This document owns the means. There is no separate target doc: section 1 holds the ends until the
operator rules on section 3. After that, Self moves the rulings and invariants into
`contracts/cultnet/cultnet-distributed-database.md` ("Error Codes") and the schema's own description.
Open: operator questions Q1-Q8 (section 3). No cut is dispatched until Q1, Q2, Q3 and Q8 are ruled. Q4-Q7
change only single emit sites and can be ruled during Cut 1.
Depends on: `hands/cultmesh-put-serve-bound` merging first. It introduces `document_unservable` in Python
(`fd394d54`), a codeless unservable refusal in cultmesh-ts (`0eb7434a`) and `CultMeshRudpRejectionReason` in
cultmesh-rs (`c9cdad12`, `85288a5a`). The cuts below convert those sites. Their anchors are by commit, not by
line, because the branch is not on main.
Map branch: `main`.
---
## 0. What the Body does today (probed)
Every claim in this table was established by running code. The C#, Rust and TypeScript probes ran on
Yggdrasil through `ygg-verify.sh` at `8fc74c70`. The Python bytes came from `msgpack.packb` on the exact
dicts `cultmesh_py/server.py:573-589` and `cultmesh-ts/src/index.ts:5921-5926` build. The six probe inputs:
| Input | Bytes from |
|---|---|
| `cs_cursor_stale` | Rust `tests/error_message.rs` captured C# hex (`cursor_stale`, details `{asOf:1,current:2}`) |
| `cs_routing` | C# `new CultNetErrorMessage { Error, RoutingHint = { Reason="not_primary", Shard=... } }` serialized with `CultNetSchemaMessageSerialization` |
| `cs_codeless` | C# `new CultNetErrorMessage { Error = "boom" }` |
| `cs_unknown_code` | C# with `Code = "document_unservable"`, empty `CultNetErrorDetails` |
| `py_snapshot_byte_limit` | the Python `_error_response` shape: `messageId`, `code: "snapshot_byte_limit_exceeded"`, `details {responseBytes, maxSnapshotBytes}` |
| `ts_two_key` | the cultmesh-ts shape `{schemaVersion, error}` |
Decoded by:
| Input | C# `CultNetSchemaMessageSerialization.Deserialize` | Rust `decode_cultnet_message_from_slice` | TS `parseCultNetMessage` |
|---|---|---|---|
| `cs_cursor_stale` | ok | ok, typed | ok |
| `cs_routing` | ok | ok, **routingHint dropped** (`code: None, details: None`) | ok (untyped passthrough) |
| `cs_codeless` | ok | ok | ok |
| `cs_unknown_code` | ok (free string) | **Err: "not a recognized cultnet.error.v0 code"** | **throws: /code must be one of the allowed values** |
| `py_snapshot_byte_limit` | ok, but **details silently become an all-null object** (`responseBytes`, `maxSnapshotBytes` lost), `messageId` dropped | **Err: not a recognized code** | **throws** (missing routingHint, extra messageId, code enum, details keys) |
| `ts_two_key` | ok | ok | **throws: missing routingHint, code, details** — TypeScript cannot parse the errors TypeScript sends |
Captured C# bytes for the two shapes Rust's vector file does not hold (for Hands' reuse):
- `cs_codeless`: `85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72A4626F6F6DAB726F7574696E6748696E74C0A4636F6465C0A764657461696C73C0`
- `cs_routing`: see the probe; the shard descriptor writes ten keys, nil included.
Other findings, from reading source at `8fc74c70`:
1. **The interop CI lane has been red for 40 consecutive `main` runs.** It fails at "Verify CultNet schema
contracts stayed consolidated" (`.github/workflows/cultnet-interop.yml:67-90`). That step forbids every
`*.schema.json` outside `contracts/cultnet/`, and R-AB deliberately vendors 22 copies into
`packages/cultnet-rs/contracts/cultnet/`. Those copies are held byte-equal by
`packages/cultnet-rs/tests/schema_discovery_contracts.rs:74-99`. Nothing after that step (every
cross-runtime interop test) has run on main in that time. Run `36714768445` is the latest.
2. **The schema contradicts itself.** `contracts/cultnet/cultnet.error.schema.json:29` says the code is
"Additive: a v0 peer that does not know this field ignores it" and, in the same sentence, "a closed
enum: Rust and TypeScript refuse". The closed-enum text came from Soul pass 4d (`3f942a57`). It
described what the code did; no operator ruled it. `docs/` holds no ruling on unknown codes.
3. **Emitters that break the schema:**
- cultmesh-ts sends `{schemaVersion, error}` only, at `packages/cultmesh-ts/src/index.ts:5875-5878`,
`5885-5888`, `5913-5916` and `5922-5925`, and on the branch at `0eb7434a`.
- cultmesh-py `_error_response` (`packages/cultmesh-py/src/cultmesh_py/server.py:573-589`) adds a
top-level `messageId`, omits `routingHint`, omits `code` and `details` rather than sending nil, and
sends eight out-of-enum codes: `unsupported_schema_version` (`:161`), `malformed_document_put`
(`:358`), `unregistered_document_put` (`:365`), `malformed_document_delete` (`:386`),
`unregistered_document_delete` (`:393`), `snapshot_document_limit_exceeded` (`:558`),
`snapshot_byte_limit_exceeded` (`:567`) and `simulation_observations_disabled` (`:596`). The branch
adds a ninth, `document_unservable` (`fd394d54`).
- `packages/cultnet-py/src/cultnet_py/interop_peer.py:409-417` also sends `unsupported_schema_version`.
R-2 (`docs/cultnet-selection-cut.md:2046-2053`) required that code, but it never entered the schema
enum.
4. **A second schema authority.** `packages/cultnet-py/src/cultnet_py/schema_catalog.py:17`, `:272` and
`:357-362` publish a hand-built `cultnet.error.v0` descriptor under the canonical `$id`. It has a
different shape: `messageId`, `code` as any string, `details` open, and only `schemaVersion` and `error`
required. The same file does this for every wire message (about 350 lines). This is recorded as
follow-up F-1, not cut here.
5. **Silent refusals.** cultmesh-rs's RUDP document server returns every rejection to the local caller only
(`packages/cultmesh-rs/src/rudp_document_server.rs:470-590`, `CultMeshRudpApplicationRejection`). The
peer gets nothing and times out. On the branch this includes the new `DocumentUnservable`.
6. **Receivers drop the code.**
- Every C# receiver turns an error into `new IOException(error.Error)` or
`new InvalidOperationException(error.Error)`, so a caller cannot branch on the code. The sites:
`src/GameCult.Mesh/CultMeshBodySessions.cs:310`, `CultMeshContentSessions.cs:178/256-260`,
`CultMeshPeerExchange.cs:430/466`, `CultMeshSnapshots.cs:91/107/222-226/743` and
`CultMeshVerseDiscovery.cs:273`, plus `src/GameCult.Networking/CultNetShardReplication.cs:440/574`.
- Rust reads `Error { error, .. }` at `packages/cultmesh-rs/src/lib.rs:823`.
- TS types `CultNetErrorMessage` as `{schemaVersion, error}` only (`packages/cultnet-ts/src/contracts.ts:167-170`).
- Kotlin has no error type. A request that gets an error back fails a
`require(response.schemaVersion == ...)` (`packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt:1278`,
`:1334`), and the peer's text is lost.
7. **No correlation.** Only Python echoes `messageId`.
- `cultmesh-browser` rejects an error only when exactly one operation is pending
(`packages/cultmesh-browser/src/index.ts:724-731`); otherwise the error is dropped and the operation
times out.
- C# `CultMeshContentSessions.OnError` (`:256-260`) fails every pending request on any error.
8. **The reference is inconsistent with itself.** `src/GameCult.Networking/CultNetDatabaseServer.cs:560`
refuses a variant change with a codeless error. The subscription server refuses the same condition with
`ForVariantUnsupported` (`CultNetDatabaseSubscriptionServer.cs:211/333/366`).
9. **routingHint is never consumed.** No receiver in any runtime reads it. C# builds it at
`CultNetDatabaseServer.cs:638-650`, and Rust's decode discards it.
10. **Consumer audit** (read-only, all of `F:\Projects` except CultLib; no git in Aetheria):
- Outside CultLib, nothing sets `code`, `details` or `routingHint`, and nothing branches on a code.
- Emitters:
- Huginn: `crates/huginn-daemon/src/serve.rs:288/307/312`, Rust struct literals
`Error { error, code: None, details: None }`, pinned at cultnet-rs `8fc74c70`.
- AetheriaEve: `Aetheria.State.Daemon/Program.cs:1737/1778`, C# object initializers.
- Decoders read `.error` only: Gjallar `Program.cs:574`, Mimir
`MimirOdinMoveProofEvidenceRingProvider.cs:80`, Hermodr `hermodr-daemon.cjs:1380` (hand-rolled),
AetheriaEve's smoke test (text match against its own emit text), and Huginn's tests (text match).
- Stale vendored TS:
- `StreamPixels/vendor/CultLib` carries the current closed schema, so any change here breaks its
TS error decode until it is re-vendored.
- `Heimdall/vendor/CultLib` carries a pre-R-X schema with `additionalProperties:false` and no
`routingHint`, so it already refuses every error C# sends today.
- Odin, Idunn, Muninn, Ghostlight and voidbot: zero hits.
## 1. Ends
- **One schema.** `contracts/cultnet/cultnet.error.schema.json` is the only authority on the shape and on the
code registry. C# embeds it (`GameCult.Networking.csproj:19`), TS mirrors it at build time, and Rust
vendors a copy held byte-equal by test. Nothing else publishes a `cultnet.error.v0` shape (F-1 covers
Python).
- **Every runtime emits only schema-valid errors.** This is proven per emit path against the schema, and
byte-equal to the C# reference for every registry code.
- **Every runtime decodes every error it can receive into a typed value.** A known code becomes its typed
variant with its typed details. An unknown code, if Q1 says so, becomes a typed "other" that keeps the
code string and the raw details. No well-formed error is ever refused, and no field is silently dropped
(routingHint, messageId).
- **Code and details have one owner.** The details a message carries are decided by its code. The types in
every runtime make "cursor_stale with a field" unrepresentable. No free-standing code string plus a
grab-bag details object.
- **A refusal is an answer, not a silence** (R-2 precedent, and Q4). A server that refuses a peer's request,
and can still send, sends a coded error that correlates to the request.
- **The public surface is complete in every runtime**, per the consumer-expectation doctrine:
- a typed error value;
- a constructor per registry code;
- a typed exception or error that request APIs raise when the peer refuses, carrying the typed value.
## 2. Identity, lifecycle and authority
| Kind | What names it | Lifecycle | Who decides |
|---|---|---|---|
| Error message | `schemaVersion: cultnet.error.v0`; correlated to its request by `messageId` (Q3) | Sent once, in reply to one request or subscription. Never persisted. | The server that refused. |
| Error code | A snake_case string in the schema's registry (`$defs/knownCode`) | Added by one coordinated change: schema, C# constructor, a C#-written vector, and a typed variant in every runtime. Never renamed and never reused. A retired code stays in the registry marked retired, so an old reader still decodes it. | The operator rules meaning; Self keeps the registry. |
| Details for a code | The code | Frozen with the code. Adding a key to an existing code's details is a new code (or a v1), never an in-place edit (see Q2). | Same as the code. |
| Unknown code, at a reader | The raw string | Preserved as typed `Other`. The reader acts on the `error` text and the absence of a known code. | The reader (Q1). |
| Vector file | `contracts/cultnet/interop/error-vectors.cs-written.json` | Regenerated only by the C# writer test under `CULTNET_WRITE_VECTORS=1` and committed. Every other runtime judges it read-only. | C#, the reference. |
| routingHint | Shard authority refusal | Sent with a null code (unchanged). Typed in every decoder. | CultNetDatabaseServer's authority path. |
## 3. Operator questions
**Rulings (operator, 2026-09-30):** "Q1: B / Q2: B / Q3: A / Q8: A".
- **Q1:** closed for emitters, open for readers. An unknown code decodes to a typed `Other`.
- **Q2:** per-code details.
- **Q3:** a nullable top-level `messageId`.
- **Q8:** v0 is corrected in place, with the semver bumps the policy requires.
- Q4-Q7 are still open.
- Cut 1 waits for `hands/cultmesh-put-serve-bound` to merge.
### Q1. What does a reader do with an error code it does not know?
Today Rust and TypeScript refuse the whole message. The probes above show the result: the peer's error text
is lost and the caller sees "not a recognized code". In cultmesh-rs's snapshot client the `?` at `lib.rs:812-815`
even turns a peer refusal into a local decode failure. So under a closed enum every new code breaks every
reader that has not upgraded. By `docs/semver-policy.md:86-91` that makes each new code a breaking wire
change, and nine codes are already in flight.
- **A. Closed enum (today).** Readers refuse unknown codes. Every code addition is a coordinated,
semver-major wire change across all runtimes and consumers.
- **B. Closed for emitters, open for readers.** The schema keeps the registry, and every runtime's emit
constructors can produce only registry codes (tests prove each emit path valid against the schema).
Readers decode an unrecognized code (matching `^[a-z][a-z0-9_]*$`) into a typed `Other { code, details }`.
A new code is an additive, minor change: old readers still surface the text and know "an error I cannot
classify". This is how gRPC status codes and protobuf open enums behave.
- **C. Fully open.** No registry; any string, emit and decode.
**Recommended: B.** A refusal must never fail to arrive because it is more specific than the reader
expected. B keeps the registry as the single meaning authority. C loses the meaning authority, and A makes
the error channel the most brittle message in CultNet.
Depends on this: schema `code` shape, every decoder, the doc text at `cultnet-distributed-database.md:107-115`.
### Q2. How are details shaped on the wire?
Today C# writes all four detail keys (`field`, `value`, `asOf`, `current`), nil included, because
`CultNetErrorDetails` (`CultNetSchemaMessages.cs:592-603`) is a MessagePack map-mode object. The codes in
flight need new keys: `responseBytes`, `maxSnapshotBytes`, `documentCount`, `maxSnapshotDocuments` and
`schemaVersion`.
- **A. One shared details record; grow it.** Every detail key of every code sits on one record, written as
nil when unused. Each new key changes the bytes of every existing error that has details, so all of the
C# vectors must be recaptured on every addition. It also breaks every strict reader: TS Ajv, and
StreamPixels' vendored schema, have `additionalProperties:false`. And the type cannot tie a key to its
code.
- **B. Per-code details.** `details` carries exactly the keys its code defines. A defined-but-optional key
is present as nil; another code's keys are absent. Keys are named for their meaning, which retires the
`field: "schemaId"` / `field: "subscriptionId"` indirection that exists only because the record is
shared. Adding a code never changes an existing code's bytes. In every runtime, code and details are
fused into one tagged type:
- Rust: an enum with data.
- C#: sealed subclasses, with a custom formatter for `CultNetErrorMessage`. There is precedent in
`src/GameCult.Mesh/CultMeshRouteRecordFormatter.cs`.
- TS: a discriminated union.
- Kotlin: a sealed class.
- Python: constructors per code over a frozen dataclass.
The churn is one time. The five existing C# cases with details change bytes, and all eight Rust hex
cases move into the shared vector file anyway (Cut 1).
- **C. Opaque string map.** Untyped; no runtime can offer a typed accessor.
**Recommended: B.** A is the churn the brief warned of, and it recurs on every future code. B pays it once,
at the moment every vector is being recaptured regardless.
Depends on this: the schema's `details`, the C# model and formatter, `encode_error_details`
(`packages/cultnet-rs/src/contracts.rs:1339-1366`), and every decoder.
### Q3. Does an error carry the `messageId` of the request it refuses?
Python already sends `messageId` (`server.py:583`); the schema forbids it. Without it:
- `cultmesh-browser` drops an error whenever two operations are pending (`index.ts:724-731`);
- C# content sessions fail every pending request on any error (`CultMeshContentSessions.cs:256-260`).
Adding a top-level key changes the bytes of every C# error. That happens once, in the same recapture as Q2.
- **A.** Add a nullable top-level `messageId`. A reply to a request echoes the request's id, and an error
with no request (a subscription push, an unsolicited refusal) sends nil. Receivers correlate by it and
fall back to their current behavior when it is nil.
- **B.** No correlation. Python deletes the field.
**Recommended: A.** Correlation is a capability any multiplexing client needs, and one runtime already
depends on it.
### Q4. Do servers send `document_unservable` (and every other refusal they can) to the peer?
Self ruled that "unservable" must be distinguishable as a typed code in every runtime. On the branch:
Python sends it coded, TypeScript sends it codeless, and Rust tells only its local caller, so the Rust
peer times out. The same Rust silence covers every other document-server rejection: a sink refusal, a
failed snapshot source, too many snapshot documents, and a snapshot response that is too large.
- **A.** Every refusal of a peer's request that the server can still send is answered with a coded error
carrying the request's `messageId`, and is also reported locally (Rust's `ApplicationRejected` stays as
the local report). The refusals that cannot be answered are the ones where the answer itself cannot be
queued: Rust `PayloadBudgetFull` and `ResponseQueueFailed`. Those stay local-only, and the doc says so.
- **B.** Local-only in Rust; peer replies only where a runtime already sends them.
**Recommended: A.** R-2 already ruled that a silence is worse than a missing feature
(`docs/cultnet-selection-cut.md:2049-2053`), and the operator's ruling that CultMesh "refuses" puts only
means something to the party that sent the put.
### Q5. Does Python's `unregistered_document_put` / `unregistered_document_delete` become `unowned_schema`?
Python sends these when the node has no registered document type for the record's schemaId
(`cultmesh_py/node.py:653-666`). C# `unowned_schema` means "no shard owns this schema and key"
(`CultNetSchemaMessages.cs:537-544`). To a peer both mean "this server will not take that schema here; route
elsewhere". The put and delete variants duplicate what the peer already knows (it sent the message).
- **A.** Both collapse into `unowned_schema { schemaId }`. The registry text broadens to "this server
stores no document for that schema (and key)".
- **B.** Join as a separate `schema_unregistered { schemaId }`.
- **C.** Join both codes as they are.
**Recommended: A.** It is one meaning at the peer, and there is one action to take.
### Q6. Does Python's `malformed_document_put` / `malformed_document_delete` become one `message_invalid { field }`?
These are sent when the request lacks a required field (`server.py:355-359`, `383-387`). C# has the same
condition, codeless, at `CultNetDatabaseServer.cs:373/415` and `CultNetDatabaseSubscriptionServer.cs:148`
("requires a subscriptionId or messageId").
- **A.** One `message_invalid { field }`, naming the missing or ill-typed field. It is used by Python's four
sites and C#'s three.
- **B.** Join both per-message codes and leave C# codeless.
**Recommended: A.** The per-message suffix repeats the request's own schemaVersion, and C# gains a code for
free.
### Q7. Does `simulation_observations_disabled` become `unsupported_schema_version`?
Python sends it when the observation hub is off (`server.py:593-598`), with
`details { schemaVersion: "cultnet.simulation_observation.v0" }`, which is exactly the
`unsupported_schema_version` shape. cultmesh-ts's "no cache for snapshot requests" and "no operation
handler" refusals (`index.ts:5875`, `5913`) mean the same thing: this server does not serve that message.
- **A.** Collapse. `unsupported_schema_version` means "this server does not serve messages of this
schemaVersion: unknown, a newer version, or not enabled here". The three TS sites and Huginn's "answers
only X, Y, Z" adopt it.
- **B.** Keep a separate `feature_disabled`-style code.
**Recommended: A.** A peer cannot act differently on "not enabled" and "not known". Python's
`verify.py:335` check follows the rename.
### Q8. Is this a correction of `cultnet.error.v0` in place, or a new `cultnet.error.v1`?
Under Q2-B and Q3-A, strict v0 readers (TS Ajv and vendored copies) refuse the new bytes. Tolerant readers
(C#, Rust, Python) keep working, because the probes show they ignore unknown keys and treat a missing key
as null.
- **A. In place.** The v0 contract has never been decodable across all five runtimes (section 0). No
consumer outside CultLib reads code or details (finding 10). Every publishing package takes the semver
bump its policy requires:
- `org.gamecult.cultlib` major, because the public `CultNetErrorDetails` properties change;
- `cultnet-py` and `cultmesh-py` major, because emitted codes change;
- Rust pins move deliberately.
StreamPixels and Heimdall re-vendor.
- **B. v1.** Add `cultnet.error.v1`, keep v0 emit for peers that did not advertise v1 in hello, and run
two error encoders in every runtime until v0 retires.
**Recommended: A.** B builds a version-negotiation path for a message nobody outside CultLib parses beyond
`.error`. The one thing every old reader does need, the `error` text, survives A in every tolerant runtime.
## 4. The registry (as recommended; rewrite on each ruling)
| Code | Meaning | Details (per Q2-B; `?` = present-as-nil when absent) | Emitted by after this map |
|---|---|---|---|
| `selection_invalid` | A selection field failed the door. | `{ field, value? }` | C#, Rust (selection) |
| `cursor_stale` | The cursor predates the server's retained history. | `{ asOf, current }` | C#, Rust |
| `cursor_invalid` | The cursor could not be parsed. | none (details nil) | C#, Rust |
| `reference_outside_target` | A hop reference left the declared target. | none | C#, Rust |
| `variant_unsupported` | The selection selects a document variant CultNet does not carry. | `{ subscriptionId? }` | C# (incl. `DatabaseServer.cs:560`) |
| `unowned_schema` | This server stores no document for that schema (and key). | `{ schemaId }` | C#, Python (Q5) |
| `unsupported_schema_version` | This server does not serve messages of this schemaVersion. | `{ schemaVersion }` | Python, TS, Huginn (Q7) |
| `message_invalid` | The request lacks a required field or has an ill-typed one. | `{ field }` | C#, Python (Q6) |
| `snapshot_document_limit_exceeded` | The snapshot answer would exceed the document bound. | `{ documentCount, maxSnapshotDocuments }` | Python, Rust (Q4) |
| `snapshot_byte_limit_exceeded` | The snapshot answer would exceed the byte bound. | `{ responseBytes, maxSnapshotBytes }` | Python, Rust (Q4) |
| `document_unservable` | A put was refused because no snapshot could ever return the stored document under the byte bound. | `{ responseBytes, maxSnapshotBytes }` | Python, TS, Rust (Q4) |
A null code is legal and means "unclassified". Examples: an exception message from a failed snapshot
source, an authority refusal carrying `routingHint`, and "not authorized"
(`CultNetDatabaseSubscriptionServer.cs:165`). This map does not invent codes for those.
`snapshot_byte_limit_exceeded` and `document_unservable` have identical details. That is deliberate, and it
is the fixture trap named in the verification rules.
---
## 5. Cuts
All cuts are made on one campaign branch, `hands/error-contract`, from `main` after
`hands/cultmesh-put-serve-bound` merges. The branch merges to `main` once, after Cut 5's Soul pass. The
reason: Cut 1 changes C# bytes, and Rust's static hex cases would stay green against the stale bytes
(finding: `tests/error_message.rs` pins constants, not the reference). A partial merge would ship a
wire split while every suite passed. Each cut gets its own worktree, created by Self. Heavy verification
runs through `ygg-verify.sh`, never on Starfire.
### Cut 0. Unblock the interop lane
- **Repo/branch:** CultLib, `hands/interop-guard` from `main`, merged alone ahead of the campaign. It
does not depend on the rulings.
- **Deletes first:** nothing.
- **Per-file changes:** `.github/workflows/cultnet-interop.yml:72`. The `$stray` filter also accepts
`^packages/cultnet-rs/contracts/cultnet/`, with a comment naming R-AB and
`packages/cultnet-rs/tests/schema_discovery_contracts.rs:74-99` as the equality guard. The namespace
check is unchanged.
- **Authority map:** no ownership change. The canonical owner stays `contracts/cultnet/`, and the Rust
mirror is derived from it, with drift judged by the Rust test.
- **Verification:**
- Push the branch and run the workflow. Report every step that goes red after the guard. None of those
steps has run on main for 40 runs, so expect more red, and treat each failure as its own finding;
do not fix it in this cut.
- Negative: edit one byte of a vendored copy locally and check that the Rust equality test fails.
Reach it on Yggdrasil with `cd packages/cultnet-rs && cargo test --test schema_discovery_contracts`.
- **Ledger:** +3 lines.
### Cut 1. The contract and the C# reference
- **Repo/branch:** CultLib `hands/error-contract`, worktree named by Self.
- **First:** capture the current C# bytes for all eight factory cases, plus `cs_codeless` and `cs_routing`,
into the Hands report. This is the before-image a Soul pass compares against.
- **Deletes first:**
- `CultNetErrorDetails` (`src/GameCult.Networking/CultNetSchemaMessages.cs:588-603`, 16 lines) as a
shared record.
- The `field`/`value`/`asOf`/`current` property block and the `[MessagePackObject]`/`[Key]` attributes
on `CultNetErrorMessage` (`:497-521`). Its serialization moves to one formatter.
- `SoulP4dProbeTests.Wire_UnownedSchemaBytes` (`tests/GameCult.Networking.Tests/SoulP4dProbeTests.cs:191-199`).
It is an `[Explicit]` hex printer, superseded by the vector writer.
- The closed-enum paragraph at `contracts/cultnet/cultnet-distributed-database.md:107-115` and
the `code`/`details` descriptions in the schema (`:28-42`).
- **Adds:**
- **Schema** (`contracts/cultnet/cultnet.error.schema.json`, rewritten; the Rust mirror is re-copied
byte-equal):
- Top level: `schemaVersion`, `error` and `routingHint` as today, plus `messageId: string|null` (Q3)
and `code`. All five are required, present-as-nil (R-X is unchanged at the top level).
- `$defs/knownCode` is the registry enum from section 4.
- `code` is `null`, or a string matching `^[a-z][a-z0-9_]*$` (Q1-B).
- `details` is decided by a top-level `oneOf`, one branch per known code, each fixing `code` by
`const` and `details` by exact `properties`/`required`/`additionalProperties:false`. Two more
branches: a null code with null details, and an unknown code (`not: {$ref knownCode}`) with details
`object|null`, open.
- Use `oneOf`/`not`/`const` only. `tests/GameCult.Networking.Tests/MiniJsonSchemaValidator.cs:30-35`
implements those and **not** `if`/`then`/`allOf`, and it refuses unimplemented keywords. TS
compiles with Ajv2020 (`packages/cultnet-ts/src/contracts.ts:1,674`), which handles both.
- **C# model** (`GameCult.Networking`):
- `CultNetErrorMessage { SchemaVersion, Error, MessageId?, RoutingHint?, Refusal? }`, where
`Code => Refusal?.Code` is derived and not settable.
- `abstract CultNetRefusal { string Code }`, with one sealed subclass per registry code carrying its
typed details (`CultNetCursorStaleRefusal(ulong AsOf, ulong Current)` and so on), plus
`CultNetUnknownRefusal(string Code, IReadOnlyDictionary<string, object?>? Details)` for Q1-B.
- `CultNetErrorCodes`: string constants, one per registry code.
- `CultNetErrorMessageFormatter : IMessagePackFormatter<CultNetErrorMessage>`. It writes the five
top-level keys in order (schemaVersion, error, messageId, routingHint, code, details), and details
as exactly the refusal's keys. It reads by key name, tolerates unknown keys at both levels, and
maps an unknown code to `CultNetUnknownRefusal`. Register it through `[MessagePackFormatter]` on
the type, the way `CultMeshRouteRecordFormatter` is wired.
- Factories: keep `ForSelectionInvalid`, `ForCursor`, `ForReferenceOutsideTarget`,
`ForVariantUnsupported` and `ForUnownedSchema` as thin constructors over the refusal types, and add
one per new code. Each takes an optional `messageId`.
- `CultNetPeerErrorException : IOException`, carrying the `CultNetErrorMessage`.
- **Vector writer:** `ErrorVectorTests.WriteErrorVectors`, gated on `CULTNET_WRITE_VECTORS=1` like
`SelectionParityVectorTests.cs:460-552`. It writes `contracts/cultnet/interop/error-vectors.cs-written.json`:
one entry per registry code (both optional-key states where a key is optional), `cs_codeless`,
`cs_routing`, one unknown code with open details, and one error with `messageId` set. Each entry
holds `{ name, messagePackBase64, expected: { error, messageId, code, details, routingHint } }`,
where `expected` is a JSON projection every runtime compares its typed decode against.
- **Per-file changes** (`8fc74c70`):
- `CultNetDatabaseServer.cs:560`: `ForVariantUnsupported(refusal.Message, subscriptionId)`.
- `CultNetDatabaseServer.cs:373`, `:415` and `CultNetDatabaseSubscriptionServer.cs:148` become
`message_invalid { field: "subscriptionId" }` (Q6-A).
- Every `SendCultNet(CultNetErrorMessage...)` that answers a request passes the request's `MessageId`.
- Each receiver listed in finding 6 throws `new CultNetPeerErrorException(error)` instead of
wrapping `error.Error`. `CultMeshContentSessions.OnError` (`:256-260`) fails only the pending entry
whose id matches `error.MessageId`, and falls back to all entries only when the id is nil.
- `CultNetSchemaContractTests.cs:165-214`: the per-factory tests are replaced by one test that validates
every vector's decoded JSON against the schema with `MiniJsonSchemaValidator`, and one test that
serializes each registry factory and asserts byte equality with its committed vector.
- Rewrite `contracts/cultnet/cultnet-distributed-database.md` "Error Codes" to the registry, and state
the Q1 and Q2 rules.
- **Authority map:**
- Owner: the schema's registry owns code meaning and detail shape. `CultNetErrorMessageFormatter` owns
C# bytes. The refusal subclass owns the code-to-details binding.
- Inputs: the refusal instance, messageId, routingHint.
- Outputs: bytes, and `cs-written.json` for the other runtimes.
- Derived state: `CultNetErrorMessage.Code` becomes derived; the refusal decides it.
- Forbidden writers: any `Code = "..."` assignment (it cannot compile); `[Key]` attribute serialization
of the message; hand-typed hex anywhere.
- Shared paths: every server send uses the factories; every receiver uses `CultNetPeerErrorException`.
- Deletion line: `CultNetErrorDetails` and the settable `Code`.
- **Verification** (`dotnet` image):
- Builds: `tests/GameCult.Networking.Tests`, `tests/GameCult.Mesh.Tests`.
- `ErrorVectorTests` with `CULTNET_WRITE_VECTORS=1`, then without, and check that the committed file
reproduces byte-for-byte.
- Tests and the rules they pin:
- vector round trip: bytes, decode, bytes are equal for every entry;
- a registry factory is byte-equal to its vector;
- every vector validates against the schema;
- an unknown code decodes to `CultNetUnknownRefusal` with its raw details intact;
- a Python-shaped error (`py_snapshot_byte_limit` above, with the Q2/Q3 shape) decodes typed with
`responseBytes` and `maxSnapshotBytes` intact. Today's C# silently empties those.
- content-session correlation: two pending requests, an error for the second, and the first still
completes.
- Mutation: Stryker.NET on the formatter and factories. The survivors that must die:
- swapping two known codes whose details are identical (`snapshot_byte_limit_exceeded` and
`document_unservable`);
- padding a nil key into details;
- dropping `messageId` on encode;
- mapping an unknown code to null.
- Negative greps (checked at HEAD, where they hit the sites being cut):
- `git grep -nE 'new (IOException|InvalidOperationException)\(error\.Error\)' -- src` must return 0
(9 hits at HEAD).
- `git grep -nE 'Field = "(schemaId|subscriptionId)"' -- src` must return 0 (2 at HEAD).
- Operator: none.
- **Ledger:** about -60 C# (the details record, the settable code, the probe printer, and the per-factory
schema tests), +250 (the formatter, the refusal types, the vector writer). The schema grows by about 90
JSON lines. The net is positive, bought by Q1/Q2 typing in the reference.
### Cut 2. Rust: cultnet-rs and cultmesh-rs
- **Deletes first:**
- `CultNetErrorCode` and `CultNetErrorDetails` as separate types (`packages/cultnet-rs/src/contracts.rs:252-310`,
59 lines);
- `require_error_details` (`:1121-1146`);
- `encode_error_details` (`:1339-1366`);
- the eight hex `Case`s and the header comment in `packages/cultnet-rs/tests/error_message.rs:1-160`.
That file is rewritten to read the vector file.
- Keep `require_legacy_optional_u64` (`:1149`). Check with `git grep` for other callers first; at HEAD,
only the error details use it.
- **Adds:**
- `CultNetMessage::Error { error: String, message_id: Option<String>, routing_hint: Option<CultNetShardRoutingHint>, refusal: Option<CultNetRefusal> }`.
- `enum CultNetRefusal`, with one variant per registry code and its typed fields, plus
`Other { code: String, details: Option<rmpv::Value> }`.
- `CultNetRefusal::code(&self) -> &str`.
- A constructor `CultNetMessage::error(text)` and builder-style `with_refusal` / `with_message_id`, so
consumers stop writing struct literals. Huginn has three.
- Decode reads every top-level key by name and routes unknown codes to `Other` (Q1-B).
- Encode writes the Cut 1 key order.
- `CultNetShardRoutingHint` is typed, reusing the crate's existing shard descriptor type.
- **Per-file changes:**
- `packages/cultnet-rs/src/contracts.rs:361-368` (variant), `:1097-1115` (decode), `:1300-1336` (encode).
- `packages/cultnet-rs/src/selection.rs:1210-1255`: the four refusals build `CultNetRefusal` variants.
- `packages/cultnet-rs/src/content.rs:316`: uses `CultNetMessage::error`.
- `packages/cultmesh-rs/src/lib.rs:812-827`: a peer error becomes a typed error carrying the refusal,
instead of `anyhow!` text. Name it `CultMeshPeerError`, next to the crate's existing error types.
- `packages/cultmesh-rs/src/rudp_document_server.rs` (Q4-A): each `CultMeshRudpRejectionReason` that
can be answered sends the matching coded error, correlated by the request's message id, before
returning `ApplicationRejected`. `PayloadBudgetFull` and `ResponseQueueFailed` stay local.
- **Authority map:**
- Owner: `CultNetRefusal` owns code and details. The schema owns the registry.
- Forbidden writers: string codes (`from_wire_str` survives only inside decode); struct literals of
details.
- Shared paths: selection refusals, document-server refusals and content errors all go through the one
encoder.
- **Verification** (`rust` image, `cd packages/cultnet-rs` then `cd ../cultmesh-rs`):
- `cargo test --workspace`.
- `tests/error_message.rs` pins: every vector decodes to `expected`; encoding the decoded value
reproduces the vector's bytes; the unknown-code vector decodes to `Other`; the routing vector keeps
`routing_hint`.
- cultmesh-rs: a put over the bound makes the peer receive `document_unservable` with the request's
message id, and details equal to the computed size and the bound. Run it at two bounds so that a
constant or an offset in `maxSnapshotBytes` dies.
- A peer that sends a Python-shaped `snapshot_byte_limit_exceeded` produces a `CultMeshPeerError`, not
a decode failure.
- Mutation: `cargo mutants --in-diff` on the cut's diff. Triage every survivor by name.
- Negative: `git grep -n 'CultNetErrorDetails' -- packages` must return 0.
- **Consumer note:** Huginn's `serve.rs:288/307/312` and its tests stop compiling at the next pin bump.
The migration is mechanical (`CultNetMessage::error(...)`), and `:312` adopts `unsupported_schema_version`
(Q7). Self routes it as a Huginn change; it is not part of this branch.
### Cut 3. TypeScript: cultnet-ts, cultmesh-ts and cultmesh-browser
- **Deletes first:**
- the two-field `CultNetErrorMessage` interface (`packages/cultnet-ts/src/contracts.ts:167-170`);
- the four literal error constructions in `packages/cultmesh-ts/src/index.ts:5875-5878`, `5885-5888`,
`5913-5916` and `5922-5925`, and the branch's codeless unservable site (`0eb7434a`);
- `rejectCorrelated`'s single-pending heuristic (`packages/cultmesh-browser/src/index.ts:724-731`).
- **Adds:**
- `CultNetErrorMessage` with every field.
- `type CultNetRefusal` as a discriminated union on `code`, with an `{ code: string; details: Record<string, unknown> | null; known: false }` arm.
- `createCultNetError(text, { messageId, refusal })` as the only constructor.
- `class CultNetPeerError extends Error { message: CultNetErrorMessage }`.
- `parseCultNetMessage` keeps Ajv validation against the canonical schema; it now accepts every schema
branch, including unknown codes.
- **Per-file changes:**
- cultmesh-ts: the three "does not serve this" sites (`5875`, `5913`, `5922`) send
`unsupported_schema_version` (Q7-A); the snapshot catch (`5885`) sends a null code; the unservable site
sends `document_unservable` with details.
- cultmesh-ts `6527-6529` throws `CultNetPeerError`.
- cultmesh-browser `163`, `584` and `690`: correlate by `messageId`, falling back to today's behavior only
for a nil id.
- `packages/cultnet-ts/src/idunn-odin-presence-publisher.ts:118-120` throws `CultNetPeerError`.
- The legacy `gamecult.networking.v0` path (`contracts.ts:840-844`, `:899-900`) is unchanged. That is a
separate legacy contract.
- **Verification** (`kotlin` image, which carries Node 24; `npm install`, then build cultcache-ts before
cultnet-ts):
- A new `test/error-vectors.test.ts` pins that every vector parses and deep-equals `expected`, and that
`encode(createCultNetError(...))` is byte-equal to each registry vector.
- The cultmesh-ts document-server tests assert that each refusal parses with TypeScript's own
`parseCultNetMessage`. That is the probe's `ts_two_key` failure, pinned.
- Browser: two pending operations, an error for the second, and the first still resolves.
- Mutation: StrykerJS on the diff.
- Negative: `git grep -nE 'schemaVersion: "cultnet\.error\.v0",' -- packages/cultmesh-ts/src packages/cultmesh-browser/src`
must return 0. Exclude `cultnet-ts/src/schema-discovery.ts:302` (a descriptor) and
`contracts.ts:842` (legacy decode); both are legitimate hits for the unscoped pattern.
### Cut 4. Python: cultnet-py and cultmesh-py
- **Deletes first:**
- `_error_response` (`packages/cultmesh-py/src/cultmesh_py/server.py:573-589`);
- `unsupported_schema_version_error` (`packages/cultnet-py/src/cultnet_py/interop_peer.py:409-417`);
- every string literal code in `server.py` (lines in finding 3).
- **Adds:**
- `cultnet_py/errors.py`, holding:
- `CultNetErrorCode(StrEnum)`, the registry;
- a frozen `CultNetError` dataclass (`error`, `message_id`, `routing_hint`, `code: CultNetErrorCode | str | None`,
`details: Mapping | None`);
- one constructor function per code, taking exactly that code's keys;
- `to_wire()`, which writes the five top-level keys in the Cut 1 order, nil included;
- `parse_cultnet_error(dict)`.
- `CultNetPeerError` (`client.py:46-50`) keeps `.response` and gains `.error: CultNetError`.
- **Per-file changes:**
- Sites `server.py:158-163`, `355-394`, `555-569` and `593-598` and the branch's `document_unservable`
site use the constructors, with the Q5-A/Q6-A/Q7-A codes.
- `interop_peer.py` uses the constructor.
- `verify.py:335` reads `error.error.code`.
- Tests `packages/cultmesh-py/tests/test_cultmesh.py:2088`, `3791`, `4323` and `4363-4371`, and
`packages/cultnet-py/tests/test_cultnet.py:2342`, follow the renames.
- **Verification** (the image the campaign already uses for Python; state it in the report):
- A vector test pins that every vector parses to `expected` and that each constructor's `msgpack.packb`
is byte-equal to its vector.
- Mutation: mutmut on `errors.py` and the changed server sites.
- Negative: `git grep -nE '"schemaVersion": "cultnet\.error\.v0"' -- packages/cultnet-py/src packages/cultmesh-py/src`
returns only `errors.py`.
- **Not here:** F-1 (`schema_catalog.py` publishes its own shapes).
### Cut 5. Kotlin: cultmesh-kotlin
- **Adds:**
- A `CultNetErrorMessage` data class and a `sealed class CultNetRefusal` (known codes plus `Other`),
with a `CultNetMessage.asError(): CultNetErrorMessage?` view.
- `class CultNetPeerException(val error: CultNetErrorMessage) : IOException`.
- At `CultMesh.kt:1278` and `:1334`, and at every `receiveSchemaMessage` consumer that expects a
response, an error response throws `CultNetPeerException` before the `require`.
- **Verification** (`kotlin` image, `pwsh -File build.ps1`):
- A vector test pins that every vector decodes to `expected`.
- Kotlin emits no errors today, so encode parity applies only if a Kotlin server path is added. Say
so in the report; do not invent one.
- There is no mutation tool. Behavioural tests cover it: an error reply to a schema-catalog request
raises `CultNetPeerException` carrying the code.
### Cut 6. Merge, release, consumers
- Soul runs over the whole branch before the merge:
- rerun every vector suite;
- decode each runtime's live emissions with every other runtime. The interop lane from Cut 0 is the
harness; add an error round to `tests/GameCult.Networking.InteropPeer` and
`packages/cultnet-ts/test/interop` if Soul finds a runtime pair unreached.
- Release per `docs/semver-policy.md`, with a Soul pass before tags:
- `org.gamecult.cultlib` major, and the tracked Unity DLLs rebuilt, since they embed the schema;
- `cultnet-py` and `cultmesh-py` major.
- Consumers, routed by Self as separate repo changes:
- Huginn: pin bump plus Cut 2's note.
- StreamPixels and Heimdall: re-vendor. Heimdall is already broken by R-X.
- AetheriaEve, Gjallar and Mimir compile against the live CultLib path and read only `.Error`, so they
need no change. That is inferred from the audit; the next build of each confirms it.
- Hermodr: none.
## 6. Subtraction ledger
| Cut | Removed (estimate) | Added (estimate) | Formats / targets |
|---|---|---|---|
| 0 | 0 | 3 | none |
| 1 | ~60 C#, schema text | ~250 C#, ~90 schema | + `error-vectors.cs-written.json`; - `[Explicit]` hex printer |
| 2 | ~150 (types, two helpers, 160-line hex test) | ~200 | hex constants gone; vectors read from file |
| 3 | ~40 | ~120 | none |
| 4 | ~40 | ~120 (`errors.py`) | 8 ad-hoc codes → registry |
| 5 | 0 | ~80 | none |
The net is positive, by roughly 500 lines, across five runtimes. It buys typed decode where there was
none (Kotlin, TS and Python), a correlation capability, and one vector file in place of hand-copied hex.
The pressure point: if Cut 1's formatter passes 150 lines, report it rather than trimming typing.
## 7. Follow-ups outside this map
- **F-1.** `packages/cultnet-py/src/cultnet_py/schema_catalog.py` publishes about 350 lines of hand-built
schemas under canonical `$id`s. Python should package and serve `contracts/cultnet/*.schema.json`, the
way Rust vendors them. It is a separate cut, because it covers every message, not errors. It can wait
because the catalog is advisory: no runtime validates a peer's message against a published descriptor.
- **F-2.** Codes for the remaining codeless refusals (authorization, failed snapshot sources, routing).
Wait for a consumer that branches on them.
- **F-3.** Whatever the Cut 0 lane reveals once its later steps run again.

28
docs/cultnet-rudp-ack-cut.md
View file

@ -1584,3 +1584,31 @@ profiles.
detection; the random default closes it. Document on the option.
- Repeated failed attempts walk the sequence up 4095 per 3 s and clamp at `u32::MAX−1` after ~18-27 days on one session;
make exhaustion a hard error the caller sees (the clamp removal mutant survives).
- **Cuts 3 and D Soul pass (Self, 2026-09-30).** Cut 3: the 4 MiB bound counts every byte sent above the lowest
unacknowledged sequence, acked or not (the spec's wording; the first implementation counted only pending packets).
Cut D: a send failure is either permanent (the datagram can never be sent as built, e.g. `EMSGSIZE`) or transient;
a permanent failure is returned from a caller-directed send and, inside a poll, ends only that peer's session with a
typed reason; transient failures stay counted losses. Windows `ConnectionResetError` on receive is idle, never fatal,
in every server. Cut D's negative grep `send_packet(...)?;` is obsolete: the `?` now propagates encode errors only.
- **Cut 3 merged at `1210fdc7` (2026-09-30).**
- Batch 2 fixed Rust `send`, which checked the window before purging expired packets. It also added a
new-generation test in all four runtimes.
- Soul pass on batch 2: nothing found. 18 mutants were killed. Interop was 27/27, run in `ack1d-interop:2`,
whose Linux wrappers let the Windows-only Kotlin build run.
- Residual: `packages/cultmesh-rs/Cargo.lock` is stale.
- Also recorded: `packages/cultmesh-kotlin/build.ps1` is Windows-only, so Kotlin interop runs off Windows only
through an image with wrappers.
- P2 (the expiry hole) stays with Cut F.
- **Cut D batch 2, Soul pass 2: fix first.**
- cultnet-ts emits transient send failures as `"error"`, which crashes processes with no listener. Section 10's
claim that "Node reports send failures on the socket's error event" is false on Node 24 without a callback:
they are dropped silently.
- C# stale peer objects write into the replacing session.
- cultmesh-py dies on a non-ValueError per-peer failure.
- Six guards are unpinned.
- Batch 3 is in Hands.
- **K1 gains:** Kotlin's idle resend loop (`CultMesh.kt:4033-4036`) is one `try`, so one peer's permanent failure
stops resends for every later peer. That is in addition to a refused packet not ending that peer's session.
- **Recorded (liveness, not fixed):** a lowest reliable packet lost forever stalls the sender 1,023 sequences ahead
while the peer stays alive; session timeout doesn't end it. TCP ends such a connection after a retransmission
limit (R2). Decide with Cut F (abandonment) whether a reliable packet outstanding past a bound ends the session.

9
docs/cultnet-transport-parity.md
View file

@ -271,6 +271,15 @@ Current progress:
flush where the runtime supports one, so callers that require confirmed
transport delivery do not substitute a fixed sleep. TCP-framed profiles do
not publish the field because they do not own the RUDP pending-reliable queue.
- The receive window is a contract, not a discovery. A receiver must hold 1,023
sequences and 4 MiB of payload beyond its contiguous received watermark, and
a conforming sender never exceeds either. The sender's flow window enforces
it without a wire field: a reliable packet is admitted only while its
sequence is at most 1,023 above the lowest unacked sequence and the payload
above that sequence stays within 4 MiB. The first packet is always
admissible, so a payload larger than 4 MiB goes out alone, and packets wait
in the FIFO in order, so a small packet never overtakes a large one. C#,
Rust, TypeScript, and Python enforce it; Kotlin does not yet.
- TypeScript, C#, Rust, Python, and Kotlin now share a portable
`cultnet.reconnect_policy.v0` document and deterministic exponential delay
helper, and RUDP transport profiles advertise that policy under

28
docs/document-variants-cut.md
View file

@ -369,6 +369,34 @@ stripped (probe). Other findings:
refuses v3 also contains C2a, so no released reader strips ids. C0 and C1 are never released on their own.
Raw and cross-runtime writers that cannot see ids stay sticky-only. Fix batch 4 is in Hands.
**C2a F2 ruled (operator, 2026-09-30, accepting Self's recommendation B): records nobody claims are carried
untouched.** A store may hold records under a schema id that no local type owns or lists as compatible, for
example after a rename that declared nothing.
- Both store kinds carry those records byte for byte under their own id. The catalog lists them as foreign
records.
- Writes to other records proceed.
- A write that would overwrite or relabel an unclaimed record is refused with a typed schema conflict.
- The single-file store's whole-view write stops relabelling such records onto local types. The directory
store stops refusing every commit while it holds one.
- Why: batches 8-9 showed that the danger is rewriting a record under a new identity (F1, F6), not an
undeclared old id. Batch 10's owner-beats-lister rule makes declaring an old id cheap anyway.
- Self's first lean, "refuse loudly" (A), is superseded.
**C2a batches 9-10 (Soul pass on `fdb8a60a`, 2026-09-30): hold.**
- **F6, A-B-A lost update on the single-file store.** Ruled by Self: any write that changes a record's bytes or
schema id mints a later `storedAt`.
- **F8, batch 9 over-refused.** An owner beats a lister, so v1 and v2 coexist.
- **F9.** Compatible ids compare as a set.
- **F10.** Typed refusal errors in TS and Python.
- Batch 10 is in Hands.
- **F7 ruled (operator, 2026-09-30): retire `AsSchemaAlias`.** "Schema aliasing sounds like a really bad idea, I
support retiring it."
- C# refuses a second type for one schema, as TS, Python and Rust already do. There is one type per schema in
every runtime.
- The API, its registry path (`IsExactWireAlias`) and the tests that exercised it (7 Mesh, 2 Caching) are
deleted or rewritten.
- No consumer called it.
**Operator rulings, 2026-09-30:**
- **Duplicate unique-index values do not load** ("Duplicate indices should not load"). This covers
plain records too, so R6-for-indexes becomes one general rule:

61
packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt
View file

@ -11,6 +11,7 @@ import java.io.InputStream
import java.io.OutputStream
import java.net.DatagramPacket
import java.net.DatagramSocket
import java.net.Inet6Address
import java.net.InetAddress
import java.net.InetSocketAddress
import java.net.ServerSocket
@ -581,7 +582,7 @@ object CultMesh {
connectionId: Long,
remoteHost: String,
remotePort: Int,
bindHost: String = "127.0.0.1",
bindHost: String? = null,
bindPort: Int = 0,
tuning: CultNetRudpSocketTuning = CultNetRudpSocketTuning(),
): CultNetRudpSocketTransportConnection = cultNetRudpClient(
@ -600,7 +601,7 @@ object CultMesh {
runtimeId: String,
connectionId: Long,
endpoint: CultNetRudpEndpoint,
bindHost: String = "127.0.0.1",
bindHost: String? = null,
bindPort: Int = 0,
tuning: CultNetRudpSocketTuning = CultNetRudpSocketTuning(),
): CultNetRudpSocketTransportConnection = createRudpClient(
@ -617,7 +618,7 @@ object CultMesh {
runtimeId: String,
connectionId: Long,
endpoint: String,
bindHost: String = "127.0.0.1",
bindHost: String? = null,
bindPort: Int = 0,
tuning: CultNetRudpSocketTuning = CultNetRudpSocketTuning(),
): CultNetRudpSocketTransportConnection = createRudpClient(
@ -633,7 +634,7 @@ object CultMesh {
runtimeId: String,
connectionId: Long,
peer: CultMeshPeerCard,
bindHost: String = "127.0.0.1",
bindHost: String? = null,
bindPort: Int = 0,
tuning: CultNetRudpSocketTuning = CultNetRudpSocketTuning(),
): CultNetRudpSocketTransportConnection {
@ -651,7 +652,7 @@ object CultMesh {
role: String,
shardId: String? = null,
at: Instant = Instant.now(),
bindHost: String = "127.0.0.1",
bindHost: String? = null,
bindPort: Int = 0,
tuning: CultNetRudpSocketTuning = CultNetRudpSocketTuning(),
): CultNetRudpSocketTransportConnection {
@ -664,7 +665,7 @@ object CultMesh {
runtimeId: String,
connectionId: Long,
endpoint: CultNetRudpEndpoint,
bindHost: String = "127.0.0.1",
bindHost: String? = null,
bindPort: Int = 0,
tuning: CultNetRudpSocketTuning = CultNetRudpSocketTuning(),
connectPayload: ByteArray = ByteArray(0),
@ -683,7 +684,7 @@ object CultMesh {
runtimeId: String,
connectionId: Long,
endpoint: String,
bindHost: String = "127.0.0.1",
bindHost: String? = null,
bindPort: Int = 0,
tuning: CultNetRudpSocketTuning = CultNetRudpSocketTuning(),
connectPayload: ByteArray = ByteArray(0),
@ -705,7 +706,7 @@ object CultMesh {
runtimeId: String,
connectionId: Long,
peer: CultMeshPeerCard,
bindHost: String = "127.0.0.1",
bindHost: String? = null,
bindPort: Int = 0,
tuning: CultNetRudpSocketTuning = CultNetRudpSocketTuning(),
connectPayload: ByteArray = ByteArray(0),
@ -736,7 +737,7 @@ object CultMesh {
role: String,
shardId: String? = null,
at: Instant = Instant.now(),
bindHost: String = "127.0.0.1",
bindHost: String? = null,
bindPort: Int = 0,
tuning: CultNetRudpSocketTuning = CultNetRudpSocketTuning(),
connectPayload: ByteArray = ByteArray(0),
@ -3028,22 +3029,36 @@ fun cultNetRudpServer(
)
}
/**
* The address an RUDP client binds when its caller names none: loopback for a
* loopback endpoint, otherwise the unspecified address of the endpoint's family.
* A socket bound to loopback cannot send off the host on Windows.
*/
fun cultNetRudpClientBindAddress(remote: InetAddress): InetAddress = when {
remote is Inet6Address && remote.isLoopbackAddress -> InetAddress.getByName("::1")
remote is Inet6Address -> InetAddress.getByName("::")
remote.isLoopbackAddress -> InetAddress.getByName("127.0.0.1")
else -> InetAddress.getByName("0.0.0.0")
}
fun cultNetRudpClient(
runtimeId: String,
connectionId: Long,
remoteHost: String,
remotePort: Int,
bindHost: String = "127.0.0.1",
bindHost: String? = null,
bindPort: Int = 0,
tuning: CultNetRudpSocketTuning = CultNetRudpSocketTuning(),
): CultNetRudpSocketTransportConnection {
val socket = DatagramSocket(bindPort, InetAddress.getByName(bindHost)).also { it.soTimeout = 20 }
val remote = InetAddress.getByName(remoteHost)
val bindAddress = bindHost?.let(InetAddress::getByName) ?: cultNetRudpClientBindAddress(remote)
val socket = DatagramSocket(bindPort, bindAddress).also { it.soTimeout = 20 }
return CultNetRudpSocketTransportConnection(
socket = socket,
mode = CultNetRudpSocketMode.Client,
runtimeId = runtimeId,
connectionId = connectionId,
remoteAddress = InetSocketAddress(InetAddress.getByName(remoteHost), remotePort),
remoteAddress = InetSocketAddress(remote, remotePort),
initialSequence = tuning.initialSequence,
resendDelayMs = tuning.resendDelayMs,
maxFragmentBytes = tuning.maxFragmentBytes,
@ -3368,6 +3383,7 @@ fun main(args: Array<String>) {
rudpSessionFragmentsAndReassemblesReliableOrderedPayloads()
rudpSessionAdvancesLargeFragmentSetsThroughBoundedReliableWindow()
rudpSocketTransportErgonomicFactoriesCarrySchemaFrames()
rudpClientBindsLoopbackOnlyForALoopbackEndpoint()
rudpSocketTransportHandshakesAndCarriesReliableOrderedSchemaFrames()
rudpSocketTransportCarriesFragmentedReliableOrderedSchemaFrames()
return
@ -5378,6 +5394,27 @@ private fun rudpSessionAdvancesLargeFragmentSetsThroughBoundedReliableWindow() {
check(oldAck.ackMask == 0L)
}
private fun rudpClientBoundHost(endpoint: String, bindHost: String? = null): InetAddress =
CultMesh.createRudpClient("kotlin-rudp-bind-client", 0x10203060L, endpoint, bindHost).use { client ->
InetAddress.getByName(client.profile.transports.single().host)
}
// A client bound to loopback cannot send to another host on Windows, so a
// client for a remote endpoint binds the unspecified address. A JVM wildcard
// socket is dual-stack and reports one wildcard for either family, so the
// family is not observable here.
private fun rudpClientBindsLoopbackOnlyForALoopbackEndpoint() {
check(rudpClientBoundHost("rudp://10.77.0.1:17872").isAnyLocalAddress)
check(rudpClientBoundHost("rudp://[2001:db8::1]:17872").isAnyLocalAddress)
check(rudpClientBoundHost("rudp://127.0.0.1:17872") == InetAddress.getByName("127.0.0.1"))
check(rudpClientBoundHost("rudp://[::1]:17872") == InetAddress.getByName("::1"))
check(rudpClientBoundHost("rudp://10.77.0.1:17872", "127.0.0.1") == InetAddress.getByName("127.0.0.1"))
check(rudpClientBoundHost("rudp://127.0.0.1:17872", "0.0.0.0").isAnyLocalAddress)
CultMesh.createRudpServer("kotlin-rudp-bind-server", 0x10203061L).use { server ->
check(InetAddress.getByName(server.profile.transports.single().host) == InetAddress.getByName("127.0.0.1"))
}
}
private fun rudpSocketTransportErgonomicFactoriesCarrySchemaFrames() {
val connectionId = 0x10203042L
CultMesh.createRudpServer(

17
packages/cultmesh-py/src/cultmesh_py/facade.py
View file

@ -18,6 +18,7 @@ from cultnet_py import (
CultNetShardCatalog,
CultNetSimulationObservationHub,
create_rudp_schema_transport,
rudp_client_bind_host,
wire_message_schema_catalog,
)
@ -188,7 +189,7 @@ class CultMesh:
connection_id: int,
endpoint: str | CultMeshRudpEndpoint,
*,
bind_host: str = "127.0.0.1",
bind_host: str | None = None,
bind_port: int = 0,
socket: socket_module.socket | None = None,
initial_sequence: int | None = None,
@ -204,7 +205,10 @@ class CultMesh:
if isinstance(endpoint, str)
else endpoint
)
transport_socket = socket or _bind_rudp_socket(bind_host, bind_port)
transport_socket = socket or _bind_rudp_socket(
bind_host if bind_host is not None else rudp_client_bind_host(parsed_endpoint.host),
bind_port,
)
return CultNetRudpSocketTransportConnection(
CultNetRudpSocketTransportOptions(
runtime_id=runtime_id,
@ -429,7 +433,7 @@ class CultMesh:
endpoint: str | CultMeshRudpEndpoint | None = None,
connection_id: int = 0x43554C54,
runtime_id: str = "cultmesh-python-rudp-client",
bind_host: str = "127.0.0.1",
bind_host: str | None = None,
bind_port: int = 0,
) -> CultNetRawClient:
if endpoint is None and host.lower().startswith("rudp://"):
@ -465,7 +469,7 @@ class CultMesh:
endpoint: str | CultMeshRudpEndpoint | None = None,
connection_id: int = 0x43554C54,
runtime_id: str = "cultmesh-python-rudp-client",
bind_host: str = "127.0.0.1",
bind_host: str | None = None,
bind_port: int = 0,
) -> CultNetRawClient:
return CultMesh.create_client(
@ -481,7 +485,10 @@ class CultMesh:
def _bind_rudp_socket(bind_host: str, bind_port: int) -> socket_module.socket:
transport_socket = socket_module.socket(socket_module.AF_INET, socket_module.SOCK_DGRAM)
transport_socket = socket_module.socket(
socket_module.AF_INET6 if ":" in bind_host else socket_module.AF_INET,
socket_module.SOCK_DGRAM,
)
transport_socket.bind((bind_host, bind_port))
transport_socket.settimeout(0.02)
return transport_socket

31
packages/cultmesh-py/tests/test_cultmesh.py
View file

@ -216,6 +216,37 @@ class CultMeshTests(unittest.TestCase):
self.assertEqual(received_messages[0]["schemaVersion"], "cultnet.schema_catalog_request.v0")
self.assertEqual(received_messages[0]["messageId"], "rudp-transport-factory")
def _rudp_client_bound_host(self, endpoint: str, bind_host: str | None = None) -> str:
client = CultMesh.create_rudp_client("python-cultmesh-bind-client", 0x10203060, endpoint, bind_host=bind_host)
try:
return client.socket.getsockname()[0]
finally:
client.socket.close()
# A client bound to loopback cannot send to another host on Windows, so a
# client for a remote endpoint binds the unspecified address of its family.
def test_cultmesh_rudp_client_for_a_remote_endpoint_binds_the_unspecified_address(self) -> None:
self.assertEqual(self._rudp_client_bound_host("rudp://10.77.0.1:17872"), "0.0.0.0")
self.assertEqual(self._rudp_client_bound_host("rudp://[2001:db8::1]:17872"), "::")
def test_cultmesh_rudp_client_for_a_loopback_endpoint_binds_loopback(self) -> None:
self.assertEqual(self._rudp_client_bound_host("rudp://127.0.0.1:17872"), "127.0.0.1")
self.assertEqual(self._rudp_client_bound_host("rudp://[::1]:17872"), "::1")
def test_cultmesh_rudp_client_binds_an_explicit_bind_host_whatever_the_endpoint(self) -> None:
self.assertEqual(self._rudp_client_bound_host("rudp://10.77.0.1:17872", "127.0.0.1"), "127.0.0.1")
self.assertEqual(self._rudp_client_bound_host("rudp://127.0.0.1:17872", "0.0.0.0"), "0.0.0.0")
def test_cultmesh_rudp_server_binds_loopback_unless_told_otherwise(self) -> None:
loopback = CultMesh.create_rudp_server("python-cultmesh-bind-server", 0x10203061)
anywhere = CultMesh.create_rudp_server("python-cultmesh-bind-server", 0x10203062, bind_host="0.0.0.0")
try:
self.assertEqual(loopback.socket.getsockname()[0], "127.0.0.1")
self.assertEqual(anywhere.socket.getsockname()[0], "0.0.0.0")
finally:
loopback.socket.close()
anywhere.socket.close()
def test_cultmesh_facade_creates_rudp_client_from_peer_endpoint(self) -> None:
connection_id = 0x10203042
server = CultMesh.create_rudp_server(

18
packages/cultmesh-ts/src/index.ts
View file

@ -14,6 +14,7 @@ import {
CultNetRudpSession,
CultNetPeer,
CultNetRudpSocketTransportConnection,
rudpClientBindHost,
CultNetSchemaCatalog,
CultNetShardCatalog,
cultNetBuiltinSchemaRegistry,
@ -4333,6 +4334,11 @@ export interface CultMeshRudpEndpoint {
}
export interface CultMeshRudpSocketOptions {
/**
* Local address to bind. Unset, a server binds loopback, and a client binds
* loopback for a loopback endpoint and the unspecified address of the
* endpoint's family otherwise, so it can reach a remote host.
*/
bindHost?: string;
bindPort?: number;
socket?: Socket;
@ -6168,7 +6174,7 @@ export class CultMesh {
options: CultMeshRudpSocketOptions = {},
): Promise<CultNetRudpSocketTransportConnection> {
requireNonEmpty(runtimeId, "runtimeId");
const socket = options.socket ?? (await bindRudpSocket(options));
const socket = options.socket ?? (await bindRudpSocket(options.bindHost ?? "127.0.0.1", options));
return new CultNetRudpSocketTransportConnection({
runtimeId,
socket,
@ -6194,7 +6200,10 @@ export class CultMesh {
requireNonEmpty(runtimeId, "runtimeId");
const parsedEndpoint =
typeof endpoint === "string" ? CultMesh.parseRudpEndpoint(endpoint) : endpoint;
const socket = options.socket ?? (await bindRudpSocket(options));
const socket = options.socket ?? (await bindRudpSocket(
options.bindHost ?? (await rudpClientBindHost(parsedEndpoint.host)),
options,
));
return new CultNetRudpSocketTransportConnection({
runtimeId,
socket,
@ -6619,9 +6628,8 @@ function toUint8Array(value: unknown): Uint8Array {
throw new Error("CultNet raw document payload was not binary.");
}
async function bindRudpSocket(options: CultMeshRudpSocketOptions): Promise<Socket> {
const socket = createSocket("udp4");
const host = options.bindHost ?? "127.0.0.1";
async function bindRudpSocket(host: string, options: CultMeshRudpSocketOptions): Promise<Socket> {
const socket = createSocket(host.includes(":") ? "udp6" : "udp4");
const port = options.bindPort ?? 0;
await new Promise<void>((resolve, reject) => {
socket.once("error", reject);

38
packages/cultmesh-ts/test/cultmesh.test.ts
View file

@ -2547,6 +2547,44 @@ test("CultMesh TS branded facade exposes schema and shard catalogs", () => {
);
});
async function rudpClientBoundHost(endpoint: string, bindHost?: string): Promise<string | undefined> {
const client = await CultMesh.createRudpClient("cultmesh-ts-bind-client", 0x10203060, endpoint, { bindHost });
try {
return client.profile.transports[0]?.host;
} finally {
client.close();
}
}
// A client bound to loopback cannot send to another host on Windows, so a
// client for a remote endpoint binds the unspecified address of its family.
test("CultMesh TS RUDP client for a remote endpoint binds the unspecified address", async () => {
assert.equal(await rudpClientBoundHost("rudp://10.77.0.1:17872"), "0.0.0.0");
assert.equal(await rudpClientBoundHost("rudp://[2001:db8::1]:17872"), "::");
});
test("CultMesh TS RUDP client for a loopback endpoint binds loopback", async () => {
assert.equal(await rudpClientBoundHost("rudp://127.0.0.1:17872"), "127.0.0.1");
assert.equal(await rudpClientBoundHost("rudp://[::1]:17872"), "::1");
});
test("CultMesh TS RUDP client binds an explicit bindHost whatever the endpoint", async () => {
assert.equal(await rudpClientBoundHost("rudp://10.77.0.1:17872", "127.0.0.1"), "127.0.0.1");
assert.equal(await rudpClientBoundHost("rudp://127.0.0.1:17872", "0.0.0.0"), "0.0.0.0");
});
test("CultMesh TS RUDP server binds loopback unless told otherwise", async () => {
const loopback = await CultMesh.createRudpServer("cultmesh-ts-bind-server", 0x10203061);
const any = await CultMesh.createRudpServer("cultmesh-ts-bind-server", 0x10203062, { bindHost: "0.0.0.0" });
try {
assert.equal(loopback.profile.transports[0]?.host, "127.0.0.1");
assert.equal(any.profile.transports[0]?.host, "0.0.0.0");
} finally {
loopback.close();
any.close();
}
});
test("CultMesh TS branded facade creates RUDP clients from peer endpoints", async () => {
const connectionId = 0x10203044;
const server = await CultMesh.createRudpServer(

2
packages/cultnet-py/src/cultnet_py/__init__.py
View file

@ -6,6 +6,7 @@ from .client import (
CultNetSchemaTransportFactory,
create_rudp_schema_transport,
create_tcp_framed_schema_transport,
rudp_client_bind_host,
)
from .cultmesh_contracts import (
PEER_EXCHANGE_REQUEST,
@ -230,6 +231,7 @@ __all__ = [
"compute_reconnect_delay_ms",
"create_reconnect_policy",
"create_rudp_schema_transport",
"rudp_client_bind_host",
"create_tcp_framed_schema_transport",
"create_rudp_transport_profile",
"create_tcp_framed_transport_profile",

25
packages/cultnet-py/src/cultnet_py/client.py
View file

@ -1,5 +1,6 @@
from __future__ import annotations
import ipaddress
import socket
import time
from collections.abc import Callable
@ -384,6 +385,22 @@ def create_tcp_framed_schema_transport(
)
def rudp_client_bind_host(endpoint_host: str) -> str:
"""Return the address an RUDP client binds when its caller names none.
Loopback for a loopback endpoint, otherwise the unspecified address of the
endpoint's family. A socket bound to loopback cannot send off the host on
Windows.
"""
family, _, _, _, address = socket.getaddrinfo(
endpoint_host.removeprefix("[").removesuffix("]"), None, type=socket.SOCK_DGRAM
)[0]
loopback = ipaddress.ip_address(address[0]).is_loopback
if family == socket.AF_INET6:
return "::1" if loopback else "::"
return "127.0.0.1" if loopback else "0.0.0.0"
def create_rudp_schema_transport(
*,
host: str,
@ -391,7 +408,7 @@ def create_rudp_schema_transport(
connection_id: int,
timeout_seconds: float = 4.0,
runtime_id: str = "cultnet-python-rudp-client",
bind_host: str = "127.0.0.1",
bind_host: str | None = None,
bind_port: int = 0,
initial_sequence: int | None = None,
resend_delay_ms: int = 25,
@ -400,7 +417,11 @@ def create_rudp_schema_transport(
max_fragment_bytes: int | None = None,
max_pending_reliable_packets: int | None = None,
) -> CultNetRudpSocketTransportConnection:
transport_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
if bind_host is None:
bind_host = rudp_client_bind_host(host)
transport_socket = socket.socket(
socket.AF_INET6 if ":" in bind_host else socket.AF_INET, socket.SOCK_DGRAM
)
transport_socket.bind((bind_host, bind_port))
transport_socket.settimeout(min(timeout_seconds, 0.02))
transport = CultNetRudpSocketTransportConnection(

58
packages/cultnet-py/src/cultnet_py/transport.py
View file

@ -465,6 +465,10 @@ def is_permanent_send_error(error: OSError) -> bool:
class CultNetRudpSession:
RELIABLE_SEND_WINDOW_PACKETS = 32
RECEIVED_SEQUENCE_WINDOW = 4_096
# Flow window: a reliable packet is admitted only while its sequence is at most FLOW_WINDOW_SEQUENCES above the
# lowest unacked one and the payload above that sequence stays within FLOW_WINDOW_BYTES.
FLOW_WINDOW_SEQUENCES = 1_023
FLOW_WINDOW_BYTES = 4 * 1024 * 1024
def __init__(self, options: CultNetRudpSessionOptions) -> None:
self.connection_id = _uint32(options.connection_id, "connection_id")
@ -505,6 +509,9 @@ class CultNetRudpSession:
self._received_sequences: set[int] = set()
self._latest_sequenced_by_channel: dict[str, int] = {}
self._pending_reliable: dict[int, _PendingReliablePacket] = {}
# Payload sizes of acknowledged reliable sequences above the lowest unacknowledged one. The receiver still
# holds those bytes behind the gap, so the flow window keeps counting them until the lowest sequence passes them.
self._acked_above_lowest: dict[int, int] = {}
self._queued_reliable: deque[CultNetRudpPacket] = deque()
# Every reliable sequence up to and including this one has been received
# since the peer state was last reset. Only the handshake seeds it: the
@ -567,6 +574,7 @@ class CultNetRudpSession:
self._awaiting_accept = False
self._generation += 1
self._pending_reliable.clear()
self._acked_above_lowest.clear()
self._queued_reliable.clear()
def reset_peer_state(self) -> None:
@ -683,7 +691,7 @@ class CultNetRudpSession:
options: CultNetRudpSendOptions | None = None,
) -> CultNetRudpPacket:
resolved = options or CultNetRudpSendOptions()
if resolved.reliable and len(self._pending_reliable) >= self.RELIABLE_SEND_WINDOW_PACKETS:
if resolved.reliable and (self._queued_reliable or not self._window_admits(self._next_sequence, len(payload))):
raise ValueError("RUDP reliable send window is full; receive acknowledgements before sending")
return self.send_many(channel_id, payload, resolved)[0]
@ -970,22 +978,39 @@ class CultNetRudpSession:
last_sent_at_ms=now_ms,
)
def _window_admits(self, sequence: int, payload_length: int) -> bool:
"""Whether a reliable packet may go on the wire now: the window has a slot, and its sequence and the
payload above the lowest unacked sequence stay inside the flow window. With nothing pending, any packet
is admissible."""
if len(self._pending_reliable) >= self.RELIABLE_SEND_WINDOW_PACKETS:
return False
if not self._pending_reliable:
return True
lowest = min(self._pending_reliable)
bytes_above = sum(
len(pending.packet.payload) for pending_sequence, pending in self._pending_reliable.items() if pending_sequence > lowest
) + sum(self._acked_above_lowest.values())
return sequence - lowest <= self.FLOW_WINDOW_SEQUENCES and bytes_above + payload_length <= self.FLOW_WINDOW_BYTES
def _admit_reliable_packets(
self,
packets: tuple[CultNetRudpPacket, ...],
now_ms: int,
) -> tuple[CultNetRudpPacket, ...]:
available = max(0, self.RELIABLE_SEND_WINDOW_PACKETS - len(self._pending_reliable))
ready = packets[:available]
for packet in ready:
self._track_reliable(packet, now_ms)
self._queued_reliable.extend(packets[available:])
return ready
ready: list[CultNetRudpPacket] = []
for packet in packets:
if not self._queued_reliable and self._window_admits(packet.sequence, len(packet.payload)):
self._track_reliable(packet, now_ms)
ready.append(packet)
else:
self._queued_reliable.append(packet)
return tuple(ready)
def _promote_queued_reliable(self, now_ms: int) -> tuple[CultNetRudpPacket, ...]:
available = max(0, self.RELIABLE_SEND_WINDOW_PACKETS - len(self._pending_reliable))
ready: list[CultNetRudpPacket] = []
while len(ready) < available and self._queued_reliable:
while self._queued_reliable and self._window_admits(
self._queued_reliable[0].sequence, len(self._queued_reliable[0].payload)
):
packet = self._queued_reliable.popleft()
self._track_reliable(packet, now_ms)
ready.append(packet)
@ -998,10 +1023,21 @@ class CultNetRudpSession:
raise ValueError("RUDP reliable send queue is full")
def _apply_acknowledgements(self, packet: CultNetRudpPacket) -> None:
self._pending_reliable.pop(packet.ack, None)
self._acknowledge(packet.ack)
for bit in range(32):
if packet.ack_mask & (1 << bit):
self._pending_reliable.pop(packet.ack - bit - 1, None)
self._acknowledge(packet.ack - bit - 1)
# Acknowledged sizes stop counting once the lowest unacknowledged sequence passes them, and all of them stop
# when nothing is unacknowledged.
lowest = min(self._pending_reliable, default=None)
self._acked_above_lowest = (
{} if lowest is None else {sequence: length for sequence, length in self._acked_above_lowest.items() if sequence > lowest}
)
def _acknowledge(self, sequence: int) -> None:
pending = self._pending_reliable.pop(sequence, None)
if pending is not None:
self._acked_above_lowest[sequence] = len(pending.packet.payload)
def _seed_received(self, sequence: int) -> None:
"""The handshake's one act on the watermark: the peer's Connect or Accept

226
packages/cultnet-py/tests/test_cultnet.py
View file

@ -30,6 +30,8 @@ from cultnet_py.interop_peer import (
)
from cultnet_py import (
compute_simulation_claim_hash,
create_rudp_schema_transport,
rudp_client_bind_host,
CultNetDatabaseChange,
CultNetRawClient,
CultNetRawDocumentRecord,
@ -899,6 +901,163 @@ class CultNetTests(unittest.TestCase):
self.assertFalse(is_permanent_send_error(OSError(code, "transient")), errno.errorcode[code])
self.assertFalse(is_permanent_send_error(OSError("no errno")))
# The sender's flow window: a reliable packet goes on the wire only while its sequence is at most 1,023 above
# the lowest unacked one and the payload above that sequence stays within 4 MiB.
FLOW_CONNECTION_ID = 0x464C4F57
MIB = 1024 * 1024
def _flow_session(self, initial_sequence: int) -> CultNetRudpSession:
return CultNetRudpSession(
CultNetRudpSessionOptions(connection_id=self.FLOW_CONNECTION_ID, initial_sequence=initial_sequence)
)
def _connected_flow_pair(
self, sender_initial_sequence: int, receiver_initial_sequence: int
) -> tuple[CultNetRudpSession, CultNetRudpSession]:
"""A sender and a receiver that have shaken hands: the receiver's watermark is seeded by the sender's
Connect, and the sender's Connect is acknowledged, so nothing is pending on the sender."""
sender = self._flow_session(sender_initial_sequence)
receiver = self._flow_session(receiver_initial_sequence)
sender.receive(receiver.accept_connect(sender.create_connect(0), 0), 0)
return sender, receiver
def _connected_flow_session(self, initial_sequence: int) -> CultNetRudpSession:
return self._connected_flow_pair(initial_sequence, 900)[0]
@staticmethod
def _flow_send(session: CultNetRudpSession, payload_bytes: int) -> tuple[CultNetRudpPacket, ...]:
return session.send_many("state", bytes(payload_bytes), CultNetRudpSendOptions(reliable=True))
def _flow_ack(self, sequence: int) -> CultNetRudpPacket:
return CultNetRudpPacket(CultNetRudpPacketType.ACK, self.FLOW_CONNECTION_ID, 0, sequence, 0, "control")
def _lose_g_and_fill_the_span(self, session: CultNetRudpSession) -> CultNetRudpPacket:
g = self._flow_send(session, 1)[0]
for offset in range(1, 1024):
self.assertEqual([p.sequence for p in self._flow_send(session, 1)], [g.sequence + offset])
session.receive(self._flow_ack(g.sequence + offset), 1)
return g
def test_cultnet_rudp_a_lost_packet_holds_the_sender_1023_sequences_ahead_and_is_still_delivered(self) -> None:
sender, receiver = self._connected_flow_pair(1, 100)
g = self._flow_send(sender, 1)[0]
admitted_after_g = 0
for _ in range(4200):
for packet in self._flow_send(sender, 1):
admitted_after_g += 1
receiver.receive(packet, 1)
sender.receive(receiver.create_ack(), 1)
self.assertEqual(admitted_after_g, 1023)
self.assertEqual(sender.queued_reliable_packet_count, 4200 - 1023)
retransmit = next(p for p in sender.due_resends(1000) if p.sequence == g.sequence)
delivered = receiver.receive(retransmit, 1000).delivered
self.assertEqual([frame.sequence for frame in delivered], [g.sequence])
promoted = sender.receive(receiver.create_ack_for_received(g.sequence), 1001).ready_to_send
self.assertEqual(promoted[0].sequence, g.sequence + 1024)
def test_cultnet_rudp_a_direct_send_past_the_span_is_refused_without_consuming_a_sequence(self) -> None:
sender = self._connected_flow_session(1)
g = self._lose_g_and_fill_the_span(sender)
with self.assertRaises(ValueError):
sender.send("state", bytes(1), CultNetRudpSendOptions(reliable=True))
self.assertEqual(sender.queued_reliable_packet_count, 0)
sender.receive(self._flow_ack(g.sequence), 2)
following = sender.send("state", bytes(1), CultNetRudpSendOptions(reliable=True))
self.assertEqual(following.sequence, g.sequence + 1024)
def test_cultnet_rudp_payload_above_the_lowest_unacked_sequence_is_bounded_by_4_mib(self) -> None:
sender = self._connected_flow_session(1)
g = self._flow_send(sender, self.MIB)[0]
# The lowest unacked packet's own bytes do not count: four more MiB fit above it.
for _ in range(4):
self.assertEqual(len(self._flow_send(sender, self.MIB)), 1)
self.assertEqual(len(self._flow_send(sender, self.MIB)), 0)
self.assertEqual(len(self._flow_send(sender, self.MIB)), 0)
self.assertEqual(sender.queued_reliable_packet_count, 2)
# Acking g moves the floor up one packet: exactly one MiB more fits.
promoted = sender.receive(self._flow_ack(g.sequence), 1).ready_to_send
self.assertEqual([p.sequence for p in promoted], [g.sequence + 5])
self.assertEqual(sender.queued_reliable_packet_count, 1)
def test_cultnet_rudp_a_packet_larger_than_the_window_goes_out_alone_and_waits_behind_anything_unacked(self) -> None:
self.assertEqual(len(self._flow_send(self._connected_flow_session(1), 5 * self.MIB)), 1)
sender = self._connected_flow_session(1)
g = self._flow_send(sender, 1)[0]
self.assertEqual(len(self._flow_send(sender, 5 * self.MIB)), 0)
promoted = sender.receive(self._flow_ack(g.sequence), 1).ready_to_send
self.assertEqual([p.sequence for p in promoted], [g.sequence + 1])
def test_cultnet_rudp_a_small_packet_does_not_overtake_a_queued_large_one(self) -> None:
sender = self._connected_flow_session(1)
g = self._flow_send(sender, 1)[0]
self.assertEqual(len(self._flow_send(sender, 3 * self.MIB)), 1)
self.assertEqual(len(self._flow_send(sender, 2 * self.MIB)), 0)
# A direct send refuses instead of queueing behind it, and takes no sequence.
with self.assertRaises(ValueError):
sender.send("state", bytes(1), CultNetRudpSendOptions(reliable=True))
self.assertEqual(sender.queued_reliable_packet_count, 1)
# One byte would fit above g, but the 2 MiB packet is ahead of it.
self.assertEqual(len(self._flow_send(sender, 1)), 0)
promoted = sender.receive(self._flow_ack(g.sequence), 1).ready_to_send
self.assertEqual([p.sequence for p in promoted], [g.sequence + 2, g.sequence + 3])
def test_cultnet_rudp_acknowledged_bytes_above_a_lost_packet_still_count_against_4_mib(self) -> None:
sender = self._connected_flow_session(1)
g = self._flow_send(sender, 1)[0]
# g is lost. Each 1 MiB frame above it is delivered and acknowledged, and the receiver still holds it behind
# the gap, so it keeps counting.
for offset in range(1, 5):
self.assertEqual([p.sequence for p in self._flow_send(sender, self.MIB)], [g.sequence + offset])
sender.receive(self._flow_ack(g.sequence + offset), 1)
self.assertEqual(len(self._flow_send(sender, self.MIB)), 0)
self.assertEqual(sender.queued_reliable_packet_count, 1)
with self.assertRaises(ValueError):
sender.send("state", bytes(self.MIB), CultNetRudpSendOptions(reliable=True))
# g arrives: nothing is held behind a gap any more, and the queue drains.
promoted = sender.receive(self._flow_ack(g.sequence), 2).ready_to_send
self.assertEqual([p.sequence for p in promoted], [g.sequence + 5])
# The bytes above the old gap no longer count: another MiB fits.
self.assertEqual([p.sequence for p in self._flow_send(sender, self.MIB)], [g.sequence + 6])
def test_cultnet_rudp_acknowledged_bytes_count_against_the_lowest_unacked_sequence_as_it_advances(self) -> None:
sender = self._connected_flow_session(1)
g = self._flow_send(sender, 1)[0]
h = self._flow_send(sender, 1)[0]
for _ in range(3):
self._flow_send(sender, self.MIB)
# g and the 3 MiB above h are acknowledged; h is lost.
for offset in (2, 3, 4, 0):
sender.receive(self._flow_ack(g.sequence + offset), 1)
self.assertEqual([p.sequence for p in self._flow_send(sender, self.MIB)], [g.sequence + 5])
# 4 MiB sit above h, acknowledged or not: one more byte does not fit.
self.assertEqual(len(self._flow_send(sender, 1)), 0)
# h arrives, the floor moves to g + 5, and the byte fits.
promoted = sender.receive(self._flow_ack(h.sequence), 2).ready_to_send
self.assertEqual([p.sequence for p in promoted], [g.sequence + 6])
def test_cultnet_rudp_a_new_generation_on_the_server_does_not_inherit_acknowledged_bytes(self) -> None:
# A server that accepts a fresh Connect may send before its Accept is acknowledged, so no acknowledgement of
# the new generation has moved the floor yet. Bytes the ended generation had acknowledged above its own lost
# packet belong to no receiver's hold any more and must not count.
server = self._flow_session(500)
accept = server.accept_connect(self._flow_session(1).create_connect(0), 0)
server.receive(self._flow_ack(accept.sequence), 0)
g = self._flow_send(server, 1)[0]
for offset in range(1, 5):
self.assertEqual([p.sequence for p in self._flow_send(server, self.MIB)], [g.sequence + offset])
server.receive(self._flow_ack(g.sequence + offset), 1)
self.assertEqual(len(self._flow_send(server, self.MIB)), 0)
next_accept = server.accept_connect(self._flow_session(200_000).create_connect(2), 2)
self.assertEqual(next_accept.packet_type, CultNetRudpPacketType.ACCEPT)
# The Accept is unacknowledged and the only thing pending: 1 MiB fits above it.
sent = server.send("state", bytes(self.MIB), CultNetRudpSendOptions(reliable=True))
self.assertEqual(sent.sequence, next_accept.sequence + 1)
def test_cultnet_rudp_socket_transport_handshakes_and_carries_reliable_ordered_schema_frames(self) -> None:
server_socket = bind_udp_socket()
client_socket = bind_udp_socket()
@ -2421,6 +2580,73 @@ class CultNetTests(unittest.TestCase):
self.assertEqual(CultNetSimulationConsensusCandidate.from_wire(candidates[0]), typed_candidates[0])
def _non_loopback_ipv4_address() -> str | None:
probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
probe.connect(("192.0.2.1", 9))
address = probe.getsockname()[0]
except OSError:
return None
finally:
probe.close()
return None if address.startswith("127.") or address == "0.0.0.0" else address
class CultNetRudpClientBindTests(unittest.TestCase):
def test_an_rudp_client_binds_loopback_only_for_a_loopback_endpoint(self) -> None:
self.assertEqual(rudp_client_bind_host("10.77.0.1"), "0.0.0.0")
self.assertEqual(rudp_client_bind_host("2001:db8::1"), "::")
self.assertEqual(rudp_client_bind_host("[2001:db8::1]"), "::")
self.assertEqual(rudp_client_bind_host("127.0.0.1"), "127.0.0.1")
self.assertEqual(rudp_client_bind_host("127.0.0.2"), "127.0.0.1")
self.assertEqual(rudp_client_bind_host("::1"), "::1")
self.assertEqual(rudp_client_bind_host("[::1]"), "::1")
# A socket bound to loopback cannot send off the host on Windows. On Linux
# it reaches a local non-loopback address, but from 127.0.0.1, so the
# source address shows which one the client bound.
def test_a_schema_transport_for_a_non_loopback_endpoint_does_not_send_from_loopback(self) -> None:
address = _non_loopback_ipv4_address()
if address is None:
self.skipTest("no non-loopback IPv4 address on this host")
listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
listener.bind(("0.0.0.0", 0))
listener.settimeout(2.0)
try:
with self.assertRaises(TimeoutError):
create_rudp_schema_transport(
host=address,
port=listener.getsockname()[1],
connection_id=0x43554C54,
timeout_seconds=0.2,
)
_, source = listener.recvfrom(65535)
self.assertEqual(source[0], address)
finally:
listener.close()
def test_a_schema_transport_binds_an_explicit_bind_host(self) -> None:
address = _non_loopback_ipv4_address()
if address is None:
self.skipTest("no non-loopback IPv4 address on this host")
listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
listener.bind(("0.0.0.0", 0))
listener.settimeout(2.0)
try:
with self.assertRaises(TimeoutError):
create_rudp_schema_transport(
host=address,
port=listener.getsockname()[1],
connection_id=0x43554C54,
timeout_seconds=0.2,
bind_host="127.0.0.1",
)
_, source = listener.recvfrom(65535)
self.assertEqual(source[0], "127.0.0.1")
finally:
listener.close()
class CultNetRudpFragmentBoundTests(unittest.TestCase):
def _pair(self) -> tuple[CultNetRudpSession, CultNetRudpSession]:
sender = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=311, initial_sequence=1))

37
packages/cultnet-rs/src/cultmesh.rs
View file

@ -3,6 +3,9 @@ use anyhow::anyhow;
use chrono::DateTime;
use chrono::Utc;
use std::collections::BTreeMap;
use std::net::IpAddr;
use std::net::Ipv4Addr;
use std::net::Ipv6Addr;
use std::net::SocketAddr;
use std::net::ToSocketAddrs;
use std::net::UdpSocket;
@ -240,7 +243,10 @@ impl CultNetRudpEndpoint {
#[derive(Clone, Debug)]
pub struct CultMeshRudpSocketOptions {
pub bind_host: String,
/// Local address to bind. Unset, a server binds loopback, and a client binds
/// loopback for a loopback endpoint and the unspecified address of the
/// endpoint's family otherwise, so it can reach a remote host.
pub bind_host: Option<String>,
pub bind_port: u16,
pub read_timeout: Option<Duration>,
pub initial_sequence: Option<u32>,
@ -253,7 +259,7 @@ pub struct CultMeshRudpSocketOptions {
impl Default for CultMeshRudpSocketOptions {
fn default() -> Self {
Self {
bind_host: "127.0.0.1".to_string(),
bind_host: None,
bind_port: 0,
read_timeout: Some(Duration::from_millis(20)),
initial_sequence: None,
@ -316,7 +322,8 @@ impl CultMesh {
connection_id: u32,
options: CultMeshRudpSocketOptions,
) -> Result<CultNetRudpSocketTransportConnection> {
let socket = bind_rudp_socket(&options)?;
let bind_host = options.bind_host.as_deref().unwrap_or("127.0.0.1");
let socket = bind_rudp_socket(bind_host, &options)?;
CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions {
media_reliable_expire_after_ms: None,
media_delivery: None,
@ -341,14 +348,19 @@ impl CultMesh {
endpoint: &CultNetRudpEndpoint,
options: CultMeshRudpSocketOptions,
) -> Result<CultNetRudpSocketTransportConnection> {
let socket = bind_rudp_socket(&options)?;
let remote_addr = endpoint.socket_addr()?;
let bind_host = match options.bind_host.as_deref() {
Some(host) => host.to_string(),
None => client_bind_ip(remote_addr).to_string(),
};
let socket = bind_rudp_socket(&bind_host, &options)?;
CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions {
media_reliable_expire_after_ms: None,
media_delivery: None,
runtime_id: runtime_id.into(),
socket,
mode: CultNetRudpSocketMode::Client,
remote_addr: Some(endpoint.socket_addr()?),
remote_addr: Some(remote_addr),
connection_id,
initial_sequence: options.initial_sequence,
resend_delay_ms: options.resend_delay_ms,
@ -503,10 +515,21 @@ pub fn cultnet_rudp_endpoint(endpoint: &str) -> Result<CultNetRudpEndpoint> {
Ok(CultNetRudpEndpoint { host, port })
}
fn bind_rudp_socket(options: &CultMeshRudpSocketOptions) -> Result<UdpSocket> {
/// A socket bound to loopback cannot send off the host on Windows, so a client
/// keeps loopback only when its peer is on loopback.
fn client_bind_ip(remote_addr: SocketAddr) -> IpAddr {
match (remote_addr.ip().is_loopback(), remote_addr) {
(true, SocketAddr::V4(_)) => IpAddr::V4(Ipv4Addr::LOCALHOST),
(true, SocketAddr::V6(_)) => IpAddr::V6(Ipv6Addr::LOCALHOST),
(false, SocketAddr::V4(_)) => IpAddr::V4(Ipv4Addr::UNSPECIFIED),
(false, SocketAddr::V6(_)) => IpAddr::V6(Ipv6Addr::UNSPECIFIED),
}
}
fn bind_rudp_socket(bind_host: &str, options: &CultMeshRudpSocketOptions) -> Result<UdpSocket> {
let socket = UdpSocket::bind(format!(
"{}:{}",
socket_host(&options.bind_host),
socket_host(bind_host),
options.bind_port
))?;
socket.set_read_timeout(options.read_timeout)?;

86
packages/cultnet-rs/src/rudp.rs
View file

@ -56,6 +56,13 @@ const RUDP_RECEIVE_AHEAD_WINDOW: u32 = 1_024;
const RUDP_CONNECT_ATTEMPT_MS: u64 = 3_000;
const RUDP_MAX_ORDERED_BUFFERED_FRAMES: usize = 1_024;
const RUDP_MAX_ORDERED_BUFFERED_BYTES: usize = 4 * 1024 * 1024;
/// The sender's flow window: a reliable packet goes on the wire only while its
/// sequence is at most this far above the lowest unacked one, and the payload
/// above that sequence stays within the byte bound. A receiver holds
/// `RUDP_RECEIVE_AHEAD_WINDOW - 1` sequences and `RUDP_MAX_ORDERED_BUFFERED_BYTES`
/// beyond its watermark, so a sender that obeys this never overruns it.
const RUDP_FLOW_WINDOW_SEQUENCES: u32 = RUDP_RECEIVE_AHEAD_WINDOW - 1;
const RUDP_FLOW_WINDOW_BYTES: usize = RUDP_MAX_ORDERED_BUFFERED_BYTES;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CultNetRudpPacketType {
@ -247,6 +254,11 @@ pub struct CultNetRudpSession {
received_sequences: BTreeSet<u32>,
latest_sequenced_by_channel: BTreeMap<String, u32>,
pending_reliable: BTreeMap<u32, PendingReliablePacket>,
/// Payload sizes of acknowledged reliable sequences above the lowest
/// unacknowledged one. The receiver still holds those bytes behind the gap,
/// so the flow window keeps counting them until the lowest sequence
/// advances past them.
acked_above_lowest: BTreeMap<u32, usize>,
queued_reliable: VecDeque<(CultNetRudpPacket, Option<u64>)>,
reliable_packets_expired: u64,
/// Every reliable sequence up to and including this one has been received
@ -287,6 +299,7 @@ impl CultNetRudpSession {
received_sequences: BTreeSet::new(),
latest_sequenced_by_channel: BTreeMap::new(),
pending_reliable: BTreeMap::new(),
acked_above_lowest: BTreeMap::new(),
queued_reliable: VecDeque::new(),
reliable_packets_expired: 0,
received_through: None,
@ -371,6 +384,7 @@ impl CultNetRudpSession {
self.awaiting_accept = false;
self.generation += 1;
self.pending_reliable.clear();
self.acked_above_lowest.clear();
self.queued_reliable.clear();
}
@ -585,8 +599,14 @@ impl CultNetRudpSession {
payload: Vec<u8>,
options: CultNetRudpSendOptions,
) -> Result<CultNetRudpPacket> {
// Measure the window as `send_many` will: without the deadline-passed
// sends it is about to reclaim.
if options.reliable {
self.purge_expired_reliable(options.now_ms);
}
if options.reliable
&& self.pending_reliable.len() >= CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS
&& (!self.queued_reliable.is_empty()
|| !self.window_admits(self.next_sequence, payload.len()))
{
return Err(anyhow!(
"RUDP reliable send window is full; receive acknowledgements before sending"
@ -1098,22 +1118,47 @@ impl CultNetRudpSession {
self.queued_reliable
.retain(|(_, expires_at_ms)| expires_at_ms.is_none_or(|at| now_ms <= at));
let after = self.pending_reliable.len() + self.queued_reliable.len();
if after != before {
self.forget_acked_below_lowest();
}
self.reliable_packets_expired = self
.reliable_packets_expired
.saturating_add((before - after) as u64);
}
/// Whether a reliable packet may go on the wire now: the window has a slot,
/// and its sequence and the payload above the lowest unacked sequence stay
/// inside the flow window. With nothing pending, any packet is admissible.
fn window_admits(&self, sequence: u32, payload_len: usize) -> bool {
if self.pending_reliable.len() >= CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS {
return false;
}
let Some((&lowest, _)) = self.pending_reliable.first_key_value() else {
return true;
};
let pending_bytes: usize = self
.pending_reliable
.iter()
.filter(|(pending_sequence, _)| **pending_sequence > lowest)
.map(|(_, pending)| pending.packet.payload.len())
.sum();
let acked_bytes: usize = self.acked_above_lowest.values().sum();
let bytes_above = pending_bytes + acked_bytes;
sequence.saturating_sub(lowest) <= RUDP_FLOW_WINDOW_SEQUENCES
&& bytes_above + payload_len <= RUDP_FLOW_WINDOW_BYTES
}
fn admit_reliable_packets(
&mut self,
packets: Vec<CultNetRudpPacket>,
now_ms: u64,
expires_at_ms: Option<u64>,
) -> Vec<CultNetRudpPacket> {
let available =
CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS.saturating_sub(self.pending_reliable.len());
let mut ready = Vec::with_capacity(available.min(packets.len()));
let mut ready = Vec::new();
for packet in packets {
if ready.len() < available {
if self.queued_reliable.is_empty()
&& self.window_admits(packet.sequence, packet.payload.len())
{
self.track_reliable(packet.clone(), now_ms, expires_at_ms);
ready.push(packet);
} else {
@ -1124,10 +1169,12 @@ impl CultNetRudpSession {
}
fn promote_queued_reliable(&mut self, now_ms: u64) -> Vec<CultNetRudpPacket> {
let available =
CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS.saturating_sub(self.pending_reliable.len());
let mut ready = Vec::with_capacity(available.min(self.queued_reliable.len()));
for _ in 0..available {
let mut ready = Vec::new();
while self
.queued_reliable
.front()
.is_some_and(|(packet, _)| self.window_admits(packet.sequence, packet.payload.len()))
{
let Some((packet, expires_at_ms)) = self.queued_reliable.pop_front() else {
break;
};
@ -1158,12 +1205,29 @@ impl CultNetRudpSession {
}
fn apply_acknowledgements(&mut self, packet: &CultNetRudpPacket) {
self.pending_reliable.remove(&packet.ack);
self.acknowledge(packet.ack);
for bit in 0..32 {
if (packet.ack_mask & (1_u32 << bit)) != 0 && packet.ack > bit {
self.pending_reliable.remove(&(packet.ack - bit - 1));
self.acknowledge(packet.ack - bit - 1);
}
}
self.forget_acked_below_lowest();
}
fn acknowledge(&mut self, sequence: u32) {
if let Some(pending) = self.pending_reliable.remove(&sequence) {
self.acked_above_lowest
.insert(sequence, pending.packet.payload.len());
}
}
/// Acknowledged sizes stop counting once the lowest unacknowledged
/// sequence passes them, and all of them stop when nothing is unacknowledged.
fn forget_acked_below_lowest(&mut self) {
match self.pending_reliable.first_key_value() {
Some((&lowest, _)) => self.acked_above_lowest.retain(|sequence, _| *sequence > lowest),
None => self.acked_above_lowest.clear(),
}
}
/// The handshake's one act on the watermark: the peer's Connect or Accept

65
packages/cultnet-rs/tests/cultnet.rs
View file

@ -1334,6 +1334,71 @@ fn rudp_socket_flush_waits_for_large_fragment_delivery() -> Result<()> {
Ok(())
}
fn rudp_client_bound_host(endpoint: &str, bind_host: Option<&str>) -> Result<String> {
let client = CultMesh::create_rudp_client_for_endpoint(
"rust-cultmesh-bind-client",
0x2030_4060,
endpoint,
CultMeshRudpSocketOptions {
bind_host: bind_host.map(str::to_string),
..CultMeshRudpSocketOptions::default()
},
)?;
Ok(client.profile.transports[0]
.host
.clone()
.expect("RUDP client profile advertises its bound host"))
}
/// A client bound to loopback cannot send to another host on Windows, so a
/// client for a remote endpoint binds the unspecified address of its family.
#[test]
fn cultmesh_rudp_client_for_a_remote_endpoint_binds_the_unspecified_address() -> Result<()> {
assert_eq!(rudp_client_bound_host("rudp://10.77.0.1:17872", None)?, "0.0.0.0");
assert_eq!(rudp_client_bound_host("rudp://[2001:db8::1]:17872", None)?, "::");
Ok(())
}
#[test]
fn cultmesh_rudp_client_for_a_loopback_endpoint_binds_loopback() -> Result<()> {
assert_eq!(rudp_client_bound_host("rudp://127.0.0.1:17872", None)?, "127.0.0.1");
assert_eq!(rudp_client_bound_host("rudp://[::1]:17872", None)?, "::1");
Ok(())
}
#[test]
fn cultmesh_rudp_client_binds_an_explicit_bind_host_whatever_the_endpoint() -> Result<()> {
assert_eq!(
rudp_client_bound_host("rudp://10.77.0.1:17872", Some("127.0.0.1"))?,
"127.0.0.1"
);
assert_eq!(
rudp_client_bound_host("rudp://127.0.0.1:17872", Some("0.0.0.0"))?,
"0.0.0.0"
);
Ok(())
}
#[test]
fn cultmesh_rudp_server_binds_loopback_unless_told_otherwise() -> Result<()> {
let server = CultMesh::create_rudp_server(
"rust-cultmesh-bind-server",
0x2030_4061,
CultMeshRudpSocketOptions::default(),
)?;
assert_eq!(server.profile.transports[0].host.as_deref(), Some("127.0.0.1"));
let server = CultMesh::create_rudp_server(
"rust-cultmesh-bind-server",
0x2030_4062,
CultMeshRudpSocketOptions {
bind_host: Some("0.0.0.0".to_string()),
..CultMeshRudpSocketOptions::default()
},
)?;
assert_eq!(server.profile.transports[0].host.as_deref(), Some("0.0.0.0"));
Ok(())
}
#[test]
fn cultmesh_facade_creates_rudp_client_from_peer_endpoint() -> Result<()> {
let connection_id = 0x2030_4050;

28
packages/cultnet-rs/tests/rudp_connect_generation.rs
View file

@ -575,9 +575,30 @@ fn a_flush_that_a_timeout_ended_names_the_timeout() -> Result<()> {
// Hold-buffer boundaries: 1,024 frames and 64 channels are held; one more is
// refused.
/// An acknowledgement that names only `sequence`.
fn ack_naming(sequence: u32) -> CultNetRudpPacket {
CultNetRudpPacket {
packet_type: CultNetRudpPacketType::Ack,
connection_id: CONNECTION_ID,
sequence: 0,
ack: sequence,
ack_mask: 0,
channel_id: "control".into(),
reliable: false,
ordered: false,
sequenced: false,
fragment_id: 0,
fragment_index: 0,
fragment_count: 0,
payload: Vec::new(),
}
}
/// A connected pair with one sequence the receiver will never see, so every
/// later ordered frame waits behind it. The sender's window is kept open by
/// acknowledging what the receiver has.
/// later ordered frame waits behind it. The sender's flow window would stop it
/// 1,023 sequences above its lowest unacked packet, so the missing one is
/// acknowledged by hand (the receiver never saw it) and the rest by what the
/// receiver has: the receiver's hold buffer is what this probes, not the sender.
fn held_behind_a_gap(
frames: usize,
channel_of: impl Fn(usize) -> String,
@ -585,7 +606,8 @@ fn held_behind_a_gap(
let mut sender = session(1);
let mut receiver = session(500);
handshake(&mut sender, &mut receiver)?;
let _missing = sender.send("gap", b"lost".to_vec(), ordered())?;
let missing = sender.send("gap", b"lost".to_vec(), ordered())?;
sender.receive(&ack_naming(missing.sequence), 1)?;
let mut held = 0;
for index in 0..frames {
let packet = sender.send(&channel_of(index), vec![0], ordered())?;

310
packages/cultnet-rs/tests/rudp_flow_window.rs Normal file
View file

@ -0,0 +1,310 @@
//! The sender's flow window: a reliable packet goes on the wire only while its
//! sequence is at most 1,023 above the lowest unacked one and the payload above
//! that sequence stays within 4 MiB. Everything else waits in order in the queue.
use anyhow::Result;
use cultnet_rs::*;
const CONNECTION_ID: u32 = 0x464c_4f57;
const MIB: usize = 1024 * 1024;
fn session(initial_sequence: u32) -> CultNetRudpSession {
CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: CONNECTION_ID,
initial_sequence: Some(initial_sequence),
resend_delay_ms: 25,
max_pending_reliable_packets: None,
})
}
/// A sender and a receiver that have shaken hands: the receiver's watermark is
/// seeded by the sender's Connect, and the sender's Connect is acknowledged, so
/// nothing is pending on the sender.
fn connected_pair(
sender_initial_sequence: u32,
receiver_initial_sequence: u32,
) -> Result<(CultNetRudpSession, CultNetRudpSession)> {
let mut sender = session(sender_initial_sequence);
let mut receiver = session(receiver_initial_sequence);
let connect = sender.create_connect(0, Vec::new())?;
let accept = receiver.accept_connect(&connect, 0, Vec::new())?;
sender.receive(&accept, 0)?;
Ok((sender, receiver))
}
fn connected(initial_sequence: u32) -> Result<CultNetRudpSession> {
Ok(connected_pair(initial_sequence, 900)?.0)
}
fn reliable() -> CultNetRudpSendOptions {
CultNetRudpSendOptions {
reliable: true,
..Default::default()
}
}
/// What `send_many` puts on the wire now; the rest of the packet waits queued.
fn send(session: &mut CultNetRudpSession, payload_bytes: usize) -> Result<Vec<CultNetRudpPacket>> {
session.send_many("state", vec![7; payload_bytes], reliable(), None)
}
fn ack_for(sequence: u32) -> CultNetRudpPacket {
CultNetRudpPacket {
packet_type: CultNetRudpPacketType::Ack,
connection_id: CONNECTION_ID,
sequence: 0,
ack: sequence,
ack_mask: 0,
channel_id: "control".into(),
reliable: false,
ordered: false,
sequenced: false,
fragment_id: 0,
fragment_index: 0,
fragment_count: 0,
payload: Vec::new(),
}
}
fn sequences(packets: &[CultNetRudpPacket]) -> Vec<u32> {
packets.iter().map(|packet| packet.sequence).collect()
}
/// Sends `g`, then acks every packet up to and including the one 1,023 above
/// it, leaving `g` the only unacked sequence. Returns `g`.
fn lose_g_and_fill_the_span(session: &mut CultNetRudpSession) -> Result<CultNetRudpPacket> {
let g = send(session, 1)?.remove(0);
for offset in 1..=1_023u32 {
let sent = send(session, 1)?;
assert_eq!(sequences(&sent), vec![g.sequence + offset]);
session.receive(&ack_for(g.sequence + offset), 1)?;
}
Ok(g)
}
#[test]
fn a_lost_packet_holds_the_sender_1023_sequences_ahead_and_is_still_delivered() -> Result<()> {
let (mut sender, mut receiver) = connected_pair(1, 100)?;
let g = send(&mut sender, 1)?.remove(0);
// g is lost. Everything after it is delivered and acked, until the sender stops.
let mut admitted_after_g = 0u32;
for _ in 0..4_200 {
for packet in send(&mut sender, 1)? {
admitted_after_g += 1;
receiver.receive(&packet, 1)?;
sender.receive(&receiver.create_ack(), 1)?;
}
}
assert_eq!(admitted_after_g, 1_023);
assert_eq!(sender.queued_reliable_packet_count(), 4_200 - 1_023);
let retransmit = sender
.due_resends(1_000)
.into_iter()
.find(|packet| packet.sequence == g.sequence)
.expect("g is still pending and due");
let delivered = receiver.receive(&retransmit, 1_000)?.delivered;
assert_eq!(delivered.len(), 1);
assert_eq!(delivered[0].sequence, g.sequence);
let promoted = sender
.receive(&receiver.create_ack_for_received(g.sequence), 1_001)?
.ready_to_send;
assert_eq!(promoted.first().map(|packet| packet.sequence), Some(g.sequence + 1_024));
Ok(())
}
#[test]
fn a_direct_send_past_the_span_is_refused_without_consuming_a_sequence() -> Result<()> {
let mut sender = connected(1)?;
let g = lose_g_and_fill_the_span(&mut sender)?;
assert!(sender.send("state", vec![7], reliable()).is_err());
assert_eq!(sender.queued_reliable_packet_count(), 0);
sender.receive(&ack_for(g.sequence), 2)?;
let next = sender.send("state", vec![7], reliable())?;
assert_eq!(next.sequence, g.sequence + 1_024);
Ok(())
}
#[test]
fn payload_above_the_lowest_unacked_sequence_is_bounded_by_4_mib() -> Result<()> {
let mut sender = connected(1)?;
let g = send(&mut sender, MIB)?.remove(0);
// The lowest unacked packet's own bytes do not count: four more MiB fit above it.
for _ in 0..4 {
assert_eq!(send(&mut sender, MIB)?.len(), 1);
}
assert!(send(&mut sender, MIB)?.is_empty());
assert!(send(&mut sender, MIB)?.is_empty());
assert_eq!(sender.queued_reliable_packet_count(), 2);
// Acking g moves the floor up one packet: exactly one MiB more fits.
let promoted = sender.receive(&ack_for(g.sequence), 1)?.ready_to_send;
assert_eq!(sequences(&promoted), vec![g.sequence + 5]);
assert_eq!(sender.queued_reliable_packet_count(), 1);
Ok(())
}
#[test]
fn a_packet_larger_than_the_window_goes_out_alone_and_waits_behind_anything_unacked() -> Result<()> {
let mut sender = connected(1)?;
assert_eq!(send(&mut sender, 5 * MIB)?.len(), 1);
let mut sender = connected(1)?;
let g = send(&mut sender, 1)?.remove(0);
assert!(send(&mut sender, 5 * MIB)?.is_empty());
let promoted = sender.receive(&ack_for(g.sequence), 1)?.ready_to_send;
assert_eq!(sequences(&promoted), vec![g.sequence + 1]);
Ok(())
}
#[test]
fn a_small_packet_does_not_overtake_a_queued_large_one() -> Result<()> {
let mut sender = connected(1)?;
let g = send(&mut sender, 1)?.remove(0);
assert_eq!(send(&mut sender, 3 * MIB)?.len(), 1);
assert!(send(&mut sender, 2 * MIB)?.is_empty());
// A direct send refuses instead of queueing behind it, and takes no sequence.
assert!(sender.send("state", vec![7], reliable()).is_err());
assert_eq!(sender.queued_reliable_packet_count(), 1);
// One byte would fit above g, but the 2 MiB packet is ahead of it.
assert!(send(&mut sender, 1)?.is_empty());
let promoted = sender.receive(&ack_for(g.sequence), 1)?.ready_to_send;
assert_eq!(sequences(&promoted), vec![g.sequence + 2, g.sequence + 3]);
Ok(())
}
#[test]
fn acknowledged_bytes_above_a_lost_packet_still_count_against_4_mib() -> Result<()> {
let mut sender = connected(1)?;
let g = send(&mut sender, 1)?.remove(0);
// g is lost. Each 1 MiB frame above it is delivered and acknowledged, and
// the receiver still holds it behind the gap, so it keeps counting.
for offset in 1..=4u32 {
let sent = send(&mut sender, MIB)?;
assert_eq!(sequences(&sent), vec![g.sequence + offset]);
sender.receive(&ack_for(g.sequence + offset), 1)?;
}
assert!(send(&mut sender, MIB)?.is_empty());
assert_eq!(sender.queued_reliable_packet_count(), 1);
assert!(sender.send("state", vec![7; MIB], reliable()).is_err());
// g arrives: nothing is held behind a gap any more, and the queue drains.
let promoted = sender.receive(&ack_for(g.sequence), 2)?.ready_to_send;
assert_eq!(sequences(&promoted), vec![g.sequence + 5]);
// The bytes above the old gap no longer count: another MiB fits.
assert_eq!(sequences(&send(&mut sender, MIB)?), vec![g.sequence + 6]);
Ok(())
}
#[test]
fn acknowledged_bytes_count_against_the_lowest_unacked_sequence_as_it_advances() -> Result<()> {
let mut sender = connected(1)?;
let g = send(&mut sender, 1)?.remove(0);
let h = send(&mut sender, 1)?.remove(0);
for _ in 0..3 {
send(&mut sender, MIB)?;
}
// g and the 3 MiB above h are acknowledged; h is lost.
for offset in [2, 3, 4, 0] {
sender.receive(&ack_for(g.sequence + offset), 1)?;
}
assert_eq!(sequences(&send(&mut sender, MIB)?), vec![g.sequence + 5]);
// 4 MiB sit above h, acknowledged or not: one more byte does not fit.
assert!(send(&mut sender, 1)?.is_empty());
// h arrives, the floor moves to g + 5, and the byte fits.
let promoted = sender.receive(&ack_for(h.sequence), 2)?.ready_to_send;
assert_eq!(sequences(&promoted), vec![g.sequence + 6]);
Ok(())
}
#[test]
fn acknowledged_bytes_stop_counting_when_the_lowest_unacked_packet_expires() -> Result<()> {
let mut sender = connected(1)?;
let expiring = CultNetRudpSendOptions {
reliable: true,
now_ms: 0,
reliable_expire_after_ms: Some(100),
..Default::default()
};
let g = sender
.send_many("state", vec![7], expiring, None)?
.remove(0);
for offset in 1..=4u32 {
assert_eq!(send(&mut sender, MIB)?.len(), 1);
sender.receive(&ack_for(g.sequence + offset), 1)?;
}
assert!(sender.send("state", vec![7; MIB], reliable()).is_err());
// g's deadline passes and it leaves the window; nothing is unacked, so the
// acknowledged bytes no longer sit above any gap.
let at = |now_ms| CultNetRudpSendOptions {
reliable: true,
now_ms,
..Default::default()
};
let first = sender.send_many("state", vec![7], at(200), None)?;
assert_eq!(first.len(), 1);
for _ in 0..4 {
assert_eq!(sender.send_many("state", vec![7; MIB], at(200), None)?.len(), 1);
}
Ok(())
}
#[test]
fn a_direct_send_is_not_refused_for_bytes_whose_lowest_packet_has_expired() -> Result<()> {
let mut sender = connected(1)?;
let expiring = CultNetRudpSendOptions {
reliable: true,
now_ms: 0,
reliable_expire_after_ms: Some(100),
..Default::default()
};
let g = sender
.send_many("state", vec![7], expiring, None)?
.remove(0);
for offset in 1..=4u32 {
assert_eq!(send(&mut sender, MIB)?.len(), 1);
sender.receive(&ack_for(g.sequence + offset), 1)?;
}
assert!(sender.send("state", vec![7; MIB], reliable()).is_err());
// The first call after g's deadline is a direct send: it measures the
// window without g, as `send_many` would, and goes on the wire.
let at_200 = CultNetRudpSendOptions {
reliable: true,
now_ms: 200,
..Default::default()
};
let sent = sender.send("state", vec![7; MIB], at_200)?;
assert_eq!(sent.sequence, g.sequence + 5);
assert_eq!(sender.outstanding_reliable_packet_count(), 1);
Ok(())
}
/// A server that accepts a fresh Connect may send before its Accept is
/// acknowledged, so no acknowledgement of the new generation has moved the
/// floor yet. Bytes the ended generation had acknowledged above its own lost
/// packet belong to no receiver's hold any more and must not count.
#[test]
fn a_new_generation_on_the_server_does_not_inherit_acknowledged_bytes() -> Result<()> {
let mut server = session(500);
let mut first_client = session(1);
let connect = first_client.create_connect(0, Vec::new())?;
let accept = server.accept_connect(&connect, 0, Vec::new())?;
server.receive(&ack_for(accept.sequence), 0)?;
let g = send(&mut server, 1)?.remove(0);
for offset in 1..=4u32 {
assert_eq!(sequences(&send(&mut server, MIB)?), vec![g.sequence + offset]);
server.receive(&ack_for(g.sequence + offset), 1)?;
}
assert!(send(&mut server, MIB)?.is_empty());
let mut second_client = session(200_000);
let connect = second_client.create_connect(2, Vec::new())?;
let accept = server.accept_connect(&connect, 2, Vec::new())?;
assert_eq!(accept.packet_type, CultNetRudpPacketType::Accept);
// The Accept is unacknowledged and the only thing pending: 1 MiB fits above it.
let sent = server.send("state", vec![7; MIB], reliable())?;
assert_eq!(sent.sequence, accept.sequence + 1);
Ok(())
}

1
packages/cultnet-ts/src/index.ts
View file

@ -108,6 +108,7 @@ export {
encodeRudpPacket,
isPermanentSendError,
sendRudpDatagram,
rudpClientBindHost,
CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS,
CultNetRudpReconnectLoop,
CultNetRudpSession,

6
packages/cultnet-ts/src/operation-service.ts
View file

@ -14,6 +14,7 @@ import {
encodeRudpPacket,
isPermanentSendError,
sendRudpDatagram,
rudpClientBindHost,
type CultNetRudpPacket,
} from "./rudp";
import { CultNetPeer } from "./peer";
@ -137,8 +138,9 @@ export async function invokeCultNetOperation(
options: CultNetOperationClientOptions,
): Promise<CultNetOperationResponseMessage> {
const target = parseRudpEndpoint(endpoint);
const socket = createSocket("udp4");
await bindSocket(socket, 0, "127.0.0.1");
const bindHost = await rudpClientBindHost(target.host);
const socket = createSocket(bindHost.includes(":") ? "udp6" : "udp4");
await bindSocket(socket, 0, bindHost);
const transport = new CultNetRudpSocketTransportConnection({
mode: "client",
runtimeId: options.runtimeId,

90
packages/cultnet-ts/src/rudp.ts
View file

@ -1,8 +1,22 @@
import { randomInt } from "node:crypto";
import { EventEmitter } from "node:events";
import { type RemoteInfo, type Socket } from "node:dgram";
import { lookup } from "node:dns/promises";
import type { CultNetTransportProfile } from "./contracts";
/**
* The address an RUDP client binds when its caller names none: loopback for a
* loopback endpoint, otherwise the unspecified address of the endpoint's family.
* A socket bound to loopback cannot send off the host on Windows.
*/
export async function rudpClientBindHost(endpointHost: string): Promise<string> {
const { address, family } = await lookup(endpointHost.replace(/^\[(.*)\]$/, "$1"));
if (family === 6) {
return address === "::1" ? "::1" : "::";
}
return address.startsWith("127.") ? "127.0.0.1" : "0.0.0.0";
}
import {
CultNetReconnectController,
createCultNetReconnectPolicy,
@ -19,6 +33,11 @@ const RUDP_FIXED_HEADER_BYTES = 36;
const MAX_CHANNEL_ID_BYTES = 255;
export const CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS = 32;
const RUDP_RECEIVED_SEQUENCE_WINDOW = 4_096;
// Flow window: a reliable packet is admitted only while its sequence is at most RUDP_FLOW_WINDOW_SEQUENCES above the
// lowest unacked one and the payload above that sequence stays within RUDP_FLOW_WINDOW_BYTES.
const RUDP_FLOW_WINDOW_SEQUENCES = 1_023;
const RUDP_FLOW_WINDOW_BYTES = 4 * 1024 * 1024;
/// How long a client keeps retransmitting a Connect nobody answered before it
/// abandons that attempt for a fresh one. A Connect the server answers with an
/// Ack, never an Accept, is one the server judged stale (see `acceptConnect`);
@ -198,6 +217,9 @@ export class CultNetRudpSession {
readonly #receivedSequences = new Set<number>();
readonly #latestSequencedByChannel = new Map<string, number>();
readonly #pendingReliable = new Map<number, PendingReliablePacket>();
/// Payload sizes of acknowledged reliable sequences above the lowest unacknowledged one. The receiver still holds
/// those bytes behind the gap, so the flow window keeps counting them until the lowest sequence passes them.
readonly #ackedAboveLowest = new Map<number, number>();
readonly #queuedReliable: CultNetRudpPacket[] = [];
/// Every reliable sequence up to and including this one has been received
/// since the peer state was last reset. Only the handshake seeds it: the
@ -280,6 +302,7 @@ export class CultNetRudpSession {
this.#awaitingAccept = false;
this.#generation += 1;
this.#pendingReliable.clear();
this.#ackedAboveLowest.clear();
this.#queuedReliable.splice(0);
}
@ -406,7 +429,7 @@ export class CultNetRudpSession {
payload: Uint8Array,
options: { reliable?: boolean; ordered?: boolean; sequenced?: boolean; nowMs?: number } = {},
): CultNetRudpPacket {
if (options.reliable && this.#pendingReliable.size >= CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS) {
if (options.reliable && (this.#queuedReliable.length > 0 || !this.#windowAdmits(this.#nextSequence, payload.length))) {
throw new Error("RUDP reliable send window is full; receive acknowledgements before sending.");
}
return this.sendMany(channelId, payload, options)[0]!;
@ -751,18 +774,44 @@ export class CultNetRudpSession {
});
}
/**
* Whether a reliable packet may go on the wire now: the window has a slot, and its sequence and the payload above
* the lowest unacked sequence stay inside the flow window. With nothing pending, any packet is admissible.
*/
#windowAdmits(sequence: number, payloadLength: number): boolean {
if (this.#pendingReliable.size >= CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS) return false;
if (this.#pendingReliable.size === 0) return true;
const lowest = Math.min(...this.#pendingReliable.keys());
let bytesAbove = 0;
for (const [pendingSequence, pending] of this.#pendingReliable) {
if (pendingSequence > lowest) bytesAbove += pending.packet.payload?.length ?? 0;
}
for (const length of this.#ackedAboveLowest.values()) bytesAbove += length;
return sequence - lowest <= RUDP_FLOW_WINDOW_SEQUENCES && bytesAbove + payloadLength <= RUDP_FLOW_WINDOW_BYTES;
}
#admitReliablePackets(packets: CultNetRudpPacket[], nowMs: number): CultNetRudpPacket[] {
const available = Math.max(0, CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS - this.#pendingReliable.size);
const ready = packets.slice(0, available);
for (const packet of ready) this.#trackReliable(packet, nowMs);
this.#queuedReliable.push(...packets.slice(available));
const ready: CultNetRudpPacket[] = [];
for (const packet of packets) {
if (this.#queuedReliable.length === 0 && this.#windowAdmits(packet.sequence, packet.payload?.length ?? 0)) {
this.#trackReliable(packet, nowMs);
ready.push(packet);
} else {
this.#queuedReliable.push(packet);
}
}
return ready;
}
#promoteQueuedReliable(nowMs: number): CultNetRudpPacket[] {
const available = Math.max(0, CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS - this.#pendingReliable.size);
const ready = this.#queuedReliable.splice(0, available);
for (const packet of ready) this.#trackReliable(packet, nowMs);
const ready: CultNetRudpPacket[] = [];
while (this.#queuedReliable.length > 0) {
const packet = this.#queuedReliable[0];
if (!this.#windowAdmits(packet.sequence, packet.payload?.length ?? 0)) break;
this.#queuedReliable.shift();
this.#trackReliable(packet, nowMs);
ready.push(packet);
}
return ready;
}
@ -776,12 +825,33 @@ export class CultNetRudpSession {
}
#applyAcknowledgements(packet: CultNetRudpPacket): void {
this.#pendingReliable.delete(packet.ack);
this.#acknowledge(packet.ack);
for (let bit = 0; bit < 32; bit += 1) {
if ((packet.ackMask & (1 << bit)) !== 0) {
this.#pendingReliable.delete(packet.ack - bit - 1);
this.#acknowledge(packet.ack - bit - 1);
}
}
this.#forgetAckedBelowLowest();
}
#acknowledge(sequence: number): void {
const pending = this.#pendingReliable.get(sequence);
if (pending === undefined) return;
this.#pendingReliable.delete(sequence);
this.#ackedAboveLowest.set(sequence, pending.packet.payload?.length ?? 0);
}
/// Acknowledged sizes stop counting once the lowest unacknowledged sequence passes them, and all of them stop
/// when nothing is unacknowledged.
#forgetAckedBelowLowest(): void {
if (this.#pendingReliable.size === 0) {
this.#ackedAboveLowest.clear();
return;
}
const lowest = Math.min(...this.#pendingReliable.keys());
for (const sequence of this.#ackedAboveLowest.keys()) {
if (sequence <= lowest) this.#ackedAboveLowest.delete(sequence);
}
}
/// The handshake's one act on the watermark: the peer's Connect or Accept is

205
packages/cultnet-ts/test/cultnet.test.ts
View file

@ -5,7 +5,7 @@ import { Duplex } from "node:stream";
import dgram, { type Socket } from "node:dgram";
import { rmSync, mkdtempSync } from "node:fs";
import { join } from "node:path";
import { tmpdir } from "node:os";
import { networkInterfaces, tmpdir } from "node:os";
import { z } from "zod";
import { decode as decodeMsgpack, encode } from "@msgpack/msgpack";
@ -47,6 +47,7 @@ import {
ghostlightAgentStateGeneratedContract,
parseCultNetMessage,
invokeCultNetOperation,
rudpClientBindHost,
startCultNetOperationServer,
validateGhostlightAgentStateGenerated,
validateGhostlightAgentState,
@ -1416,6 +1417,167 @@ test("rudp session advances large fragment sets through a bounded reliable windo
assert.equal(oldAck.ackMask, 0);
});
// The sender's flow window: a reliable packet goes on the wire only while its sequence is at most 1,023 above the
// lowest unacked one and the payload above that sequence stays within 4 MiB.
const FLOW_CONNECTION_ID = 0x464c4f57;
const MIB = 1024 * 1024;
function flowSession(initialSequence: number): CultNetRudpSession {
return new CultNetRudpSession({ connectionId: FLOW_CONNECTION_ID, initialSequence, resendDelayMs: 25 });
}
// A sender and a receiver that have shaken hands: the receiver's watermark is seeded by the sender's Connect, and the
// sender's Connect is acknowledged, so nothing is pending on the sender.
function connectedFlowPair(senderInitialSequence: number, receiverInitialSequence: number): [CultNetRudpSession, CultNetRudpSession] {
const sender = flowSession(senderInitialSequence);
const receiver = flowSession(receiverInitialSequence);
sender.receive(receiver.acceptConnect(sender.createConnect(0), 0), 0);
return [sender, receiver];
}
function connectedFlowSession(initialSequence: number): CultNetRudpSession {
return connectedFlowPair(initialSequence, 900)[0];
}
function flowSend(session: CultNetRudpSession, payloadBytes: number): CultNetRudpPacket[] {
return session.sendMany("state", new Uint8Array(payloadBytes), { reliable: true });
}
function flowAck(sequence: number): CultNetRudpPacket {
return { packetType: "ack", connectionId: FLOW_CONNECTION_ID, sequence: 0, ack: sequence, ackMask: 0, channelId: "control" };
}
const sequencesOf = (packets: CultNetRudpPacket[]): number[] => packets.map((packet) => packet.sequence);
function loseGAndFillTheSpan(session: CultNetRudpSession): CultNetRudpPacket {
const g = flowSend(session, 1)[0]!;
for (let offset = 1; offset <= 1023; offset++) {
assert.deepEqual(sequencesOf(flowSend(session, 1)), [g.sequence + offset]);
session.receive(flowAck(g.sequence + offset), 1);
}
return g;
}
test("a lost packet holds the sender 1023 sequences ahead and is still delivered", () => {
const [sender, receiver] = connectedFlowPair(1, 100);
const g = flowSend(sender, 1)[0]!;
let admittedAfterG = 0;
for (let index = 0; index < 4200; index++) {
for (const packet of flowSend(sender, 1)) {
admittedAfterG++;
receiver.receive(packet, 1);
sender.receive(receiver.createAck(), 1);
}
}
assert.equal(admittedAfterG, 1023);
assert.equal(sender.queuedReliablePacketCount, 4200 - 1023);
const retransmit = sender.dueResends(1000).find((packet) => packet.sequence === g.sequence)!;
const delivered = receiver.receive(retransmit, 1000).delivered;
assert.deepEqual(delivered.map((frame) => frame.sequence), [g.sequence]);
const promoted = sender.receive(receiver.createAckForReceived(g.sequence), 1001).readyToSend ?? [];
assert.equal(promoted[0]!.sequence, g.sequence + 1024);
});
test("a direct send past the span is refused without consuming a sequence", () => {
const sender = connectedFlowSession(1);
const g = loseGAndFillTheSpan(sender);
assert.throws(() => sender.send("state", new Uint8Array(1), { reliable: true }));
assert.equal(sender.queuedReliablePacketCount, 0);
sender.receive(flowAck(g.sequence), 2);
assert.equal(sender.send("state", new Uint8Array(1), { reliable: true }).sequence, g.sequence + 1024);
});
test("payload above the lowest unacked sequence is bounded by 4 MiB", () => {
const sender = connectedFlowSession(1);
const g = flowSend(sender, MIB)[0]!;
// The lowest unacked packet's own bytes do not count: four more MiB fit above it.
for (let index = 0; index < 4; index++) assert.equal(flowSend(sender, MIB).length, 1);
assert.equal(flowSend(sender, MIB).length, 0);
assert.equal(flowSend(sender, MIB).length, 0);
assert.equal(sender.queuedReliablePacketCount, 2);
// Acking g moves the floor up one packet: exactly one MiB more fits.
const promoted = sender.receive(flowAck(g.sequence), 1).readyToSend ?? [];
assert.deepEqual(sequencesOf(promoted), [g.sequence + 5]);
assert.equal(sender.queuedReliablePacketCount, 1);
});
test("a packet larger than the window goes out alone and waits behind anything unacked", () => {
assert.equal(flowSend(connectedFlowSession(1), 5 * MIB).length, 1);
const sender = connectedFlowSession(1);
const g = flowSend(sender, 1)[0]!;
assert.equal(flowSend(sender, 5 * MIB).length, 0);
assert.deepEqual(sequencesOf(sender.receive(flowAck(g.sequence), 1).readyToSend ?? []), [g.sequence + 1]);
});
test("a small packet does not overtake a queued large one", () => {
const sender = connectedFlowSession(1);
const g = flowSend(sender, 1)[0]!;
assert.equal(flowSend(sender, 3 * MIB).length, 1);
assert.equal(flowSend(sender, 2 * MIB).length, 0);
// A direct send refuses instead of queueing behind it, and takes no sequence.
assert.throws(() => sender.send("state", new Uint8Array(1), { reliable: true }));
assert.equal(sender.queuedReliablePacketCount, 1);
// One byte would fit above g, but the 2 MiB packet is ahead of it.
assert.equal(flowSend(sender, 1).length, 0);
assert.deepEqual(sequencesOf(sender.receive(flowAck(g.sequence), 1).readyToSend ?? []), [g.sequence + 2, g.sequence + 3]);
});
test("acknowledged bytes above a lost packet still count against 4 MiB", () => {
const sender = connectedFlowSession(1);
const g = flowSend(sender, 1)[0]!;
// g is lost. Each 1 MiB frame above it is delivered and acknowledged, and the receiver still holds it behind the
// gap, so it keeps counting.
for (let offset = 1; offset <= 4; offset++) {
assert.deepEqual(sequencesOf(flowSend(sender, MIB)), [g.sequence + offset]);
sender.receive(flowAck(g.sequence + offset), 1);
}
assert.equal(flowSend(sender, MIB).length, 0);
assert.equal(sender.queuedReliablePacketCount, 1);
assert.throws(() => sender.send("state", new Uint8Array(MIB), { reliable: true }));
// g arrives: nothing is held behind a gap any more, and the queue drains.
assert.deepEqual(sequencesOf(sender.receive(flowAck(g.sequence), 2).readyToSend ?? []), [g.sequence + 5]);
// The bytes above the old gap no longer count: another MiB fits.
assert.deepEqual(sequencesOf(flowSend(sender, MIB)), [g.sequence + 6]);
});
test("acknowledged bytes count against the lowest unacked sequence as it advances", () => {
const sender = connectedFlowSession(1);
const g = flowSend(sender, 1)[0]!;
const h = flowSend(sender, 1)[0]!;
for (let index = 0; index < 3; index++) flowSend(sender, MIB);
// g and the 3 MiB above h are acknowledged; h is lost.
for (const offset of [2, 3, 4, 0]) sender.receive(flowAck(g.sequence + offset), 1);
assert.deepEqual(sequencesOf(flowSend(sender, MIB)), [g.sequence + 5]);
// 4 MiB sit above h, acknowledged or not: one more byte does not fit.
assert.equal(flowSend(sender, 1).length, 0);
// h arrives, the floor moves to g + 5, and the byte fits.
assert.deepEqual(sequencesOf(sender.receive(flowAck(h.sequence), 2).readyToSend ?? []), [g.sequence + 6]);
});
// A server that accepts a fresh Connect may send before its Accept is acknowledged, so no acknowledgement of the new
// generation has moved the floor yet. Bytes the ended generation had acknowledged above its own lost packet belong to
// no receiver's hold any more and must not count.
test("a new generation on the server does not inherit acknowledged bytes", () => {
const server = flowSession(500);
const accept = server.acceptConnect(flowSession(1).createConnect(0), 0);
server.receive(flowAck(accept.sequence), 0);
const g = flowSend(server, 1)[0]!;
for (let offset = 1; offset <= 4; offset++) {
assert.deepEqual(sequencesOf(flowSend(server, MIB)), [g.sequence + offset]);
server.receive(flowAck(g.sequence + offset), 1);
}
assert.equal(flowSend(server, MIB).length, 0);
const nextAccept = server.acceptConnect(flowSession(200_000).createConnect(2), 2);
assert.equal(nextAccept.packetType, "accept");
// The Accept is unacknowledged and the only thing pending: 1 MiB fits above it.
assert.equal(server.send("state", new Uint8Array(MIB), { reliable: true }).sequence, nextAccept.sequence + 1);
});
test("CultNet contracts encode legacy bytes without Node Buffer authority", () => {
const originalBuffer = globalThis.Buffer;
try {
@ -2224,6 +2386,47 @@ test("operation service frees the slots of abandoned sessions once they idle out
}
});
test("an RUDP client binds loopback only for a loopback endpoint", async () => {
assert.equal(await rudpClientBindHost("10.77.0.1"), "0.0.0.0");
assert.equal(await rudpClientBindHost("2001:db8::1"), "::");
assert.equal(await rudpClientBindHost("[2001:db8::1]"), "::");
assert.equal(await rudpClientBindHost("127.0.0.1"), "127.0.0.1");
assert.equal(await rudpClientBindHost("127.0.0.2"), "127.0.0.1");
assert.equal(await rudpClientBindHost("::1"), "::1");
assert.equal(await rudpClientBindHost("[::1]"), "::1");
});
// A socket bound to loopback cannot send off the host on Windows. On Linux it
// reaches a local non-loopback address, but from 127.0.0.1, so the source
// address shows which one the client bound.
test("an operation client for a non-loopback endpoint does not send from loopback", async (t) => {
const address = Object.values(networkInterfaces())
.flat()
.find((entry) => entry && entry.family === "IPv4" && !entry.internal)?.address;
if (!address) {
t.skip("no non-loopback IPv4 interface on this host");
return;
}
const listener = dgram.createSocket("udp4");
await new Promise<void>((resolve) => listener.bind(0, "0.0.0.0", resolve));
const source = new Promise<string>((resolve) => listener.once("message", (_wire, remote) => resolve(remote.address)));
const request: CultNetOperationRequestMessage = {
schemaVersion: "cultnet.operation_request.v0",
messageId: "bind-host",
serviceId: "sai.vn",
operation: "describe",
payloadSchema: "gamecult.eve.plugin_abi.request.v1",
payloadEncoding: "messagepack-base64",
payload: "gaZzY2hlbWE=",
};
try {
await assert.rejects(invokeCultNetOperation(`rudp://${address}:${udpPort(listener)}`, request, { runtimeId: "bind-test", timeoutMs: 300 }));
assert.equal(await source, address);
} finally {
listener.close();
}
});
test("server-mode transport replaces its peer when a Connect arrives from another endpoint", async () => {
const serverSocket = await bindUdpSocket();
const socketA = await bindUdpSocket();

34
src/GameCult.Mesh/CultMesh.cs
View file

@ -537,8 +537,12 @@ namespace GameCult.Mesh
/// </summary>
public sealed class CultMeshRudpSocketOptions
{
/// <summary>Gets or sets the local bind host.</summary>
public string BindHost { get; set; } = "0.0.0.0";
/// <summary>
/// Gets or sets the local bind host. Unset, a server binds every IPv4 interface, and a
/// client binds loopback for a loopback endpoint and the unspecified address of the
/// endpoint's family otherwise, so it can reach a remote host.
/// </summary>
public string? BindHost { get; set; }
/// <summary>Gets or sets the local bind port.</summary>
public int BindPort { get; set; }
/// <summary>Gets or sets a caller-owned bound socket.</summary>
@ -2933,7 +2937,7 @@ namespace GameCult.Mesh
return new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
{
RuntimeId = runtimeId,
Socket = options.Socket ?? BindRudpSocket(options.BindHost, options.BindPort),
Socket = options.Socket ?? BindRudpSocket(ResolveRudpAddress(options.BindHost ?? "0.0.0.0"), options.BindPort),
Mode = CultNetRudpSocketMode.Server,
ConnectionId = connectionId,
InitialSequence = options.InitialSequence,
@ -2957,12 +2961,16 @@ namespace GameCult.Mesh
{
if (endpoint == null) throw new ArgumentNullException(nameof(endpoint));
options ??= new CultMeshRudpSocketOptions();
var remote = ResolveRudpAddress(endpoint.Host);
var bindAddress = options.BindHost == null
? RudpClientBindAddress(remote)
: ResolveRudpAddress(options.BindHost);
return new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
{
RuntimeId = runtimeId,
Socket = options.Socket ?? BindRudpSocket(options.BindHost, options.BindPort),
Socket = options.Socket ?? BindRudpSocket(bindAddress, options.BindPort),
Mode = CultNetRudpSocketMode.Client,
RemoteEndPoint = new IPEndPoint(ResolveRudpAddress(endpoint.Host), endpoint.Port),
RemoteEndPoint = new IPEndPoint(remote, endpoint.Port),
ConnectionId = connectionId,
InitialSequence = options.InitialSequence,
ResendDelayMs = options.ResendDelayMs,
@ -3234,9 +3242,21 @@ namespace GameCult.Mesh
return CultNetLocal.ConnectClient(host, port, security, configureClient);
}
private static Socket BindRudpSocket(string host, int port)
// A socket bound to loopback cannot send off the host on Windows, so a client keeps
// loopback only when its peer is on loopback.
private static IPAddress RudpClientBindAddress(IPAddress remote)
{
var ipv6 = remote.AddressFamily == AddressFamily.InterNetworkV6;
if (IPAddress.IsLoopback(remote))
{
return ipv6 ? IPAddress.IPv6Loopback : IPAddress.Loopback;
}
return ipv6 ? IPAddress.IPv6Any : IPAddress.Any;
}
private static Socket BindRudpSocket(IPAddress address, int port)
{
var address = ResolveRudpAddress(host);
var socket = new Socket(address.AddressFamily, SocketType.Dgram, ProtocolType.Udp);
socket.Bind(new IPEndPoint(address, port));
socket.ReceiveTimeout = 20;

67
src/GameCult.Networking/CultNetTransport.cs
View file

@ -893,6 +893,10 @@ namespace GameCult.Networking
{
/// <summary>Maximum reliable packets admitted to the wire before acknowledgements advance the window.</summary>
public const int ReliableSendWindowPackets = 32;
// Flow window: a reliable packet is admitted only while its sequence is at most FlowWindowSequences above the
// lowest unacked one and the payload above that sequence stays within FlowWindowBytes.
private const uint FlowWindowSequences = 1023;
private const int FlowWindowBytes = 4 * 1024 * 1024;
private const int ReceivedSequenceWindow = 4096;
// How long a client keeps retransmitting a Connect nobody answered before it abandons that
// attempt for a fresh one. A Connect the server answers with an Ack, never an Accept, is one
@ -943,6 +947,9 @@ namespace GameCult.Networking
private readonly Dictionary<string, uint> _latestSequencedByChannel = new Dictionary<string, uint>(StringComparer.Ordinal);
private readonly object _pendingReliableGate = new object();
private readonly Dictionary<uint, PendingReliablePacket> _pendingReliable = new Dictionary<uint, PendingReliablePacket>();
// Payload sizes of acknowledged reliable sequences above the lowest unacknowledged one. The receiver still
// holds those bytes behind the gap, so the flow window keeps counting them until the lowest sequence passes them.
private readonly Dictionary<uint, int> _ackedAboveLowest = new Dictionary<uint, int>();
private readonly Queue<CultNetRudpPacket> _queuedReliable = new Queue<CultNetRudpPacket>();
// Every reliable sequence up to and including this one has been received since the peer state
// was last reset. Only the handshake seeds it: the peer's Connect on the accepting side, the
@ -1164,7 +1171,7 @@ namespace GameCult.Networking
options ??= new CultNetRudpSendOptions();
lock (_pendingReliableGate)
{
if (options.Reliable && _pendingReliable.Count >= ReliableSendWindowPackets)
if (options.Reliable && (_queuedReliable.Count > 0 || !WindowAdmits(_nextSequence, payload.Length)))
throw new InvalidOperationException("RUDP reliable send window is full; receive acknowledgements before sending.");
}
return SendMany(channelId, payload, options).First();
@ -1428,6 +1435,7 @@ namespace GameCult.Networking
lock (_pendingReliableGate)
{
_pendingReliable.Clear();
_ackedAboveLowest.Clear();
_queuedReliable.Clear();
}
}
@ -1628,17 +1636,39 @@ namespace GameCult.Networking
}
}
/// <summary>
/// Whether a reliable packet may go on the wire now: the window has a slot, and its sequence and the
/// payload above the lowest unacked sequence stay inside the flow window. With nothing pending, any
/// packet is admissible. Callers hold the pending gate.
/// </summary>
private bool WindowAdmits(uint sequence, int payloadLength)
{
if (_pendingReliable.Count >= ReliableSendWindowPackets)
return false;
if (_pendingReliable.Count == 0)
return true;
var lowest = _pendingReliable.Keys.Min();
long bytesAbove = 0;
foreach (var pending in _pendingReliable)
{
if (pending.Key > lowest)
bytesAbove += pending.Value.Packet.Payload.Length;
}
foreach (var length in _ackedAboveLowest.Values)
bytesAbove += length;
return sequence - lowest <= FlowWindowSequences && bytesAbove + payloadLength <= FlowWindowBytes;
}
private IReadOnlyList<CultNetRudpPacket> AdmitReliablePackets(
IReadOnlyList<CultNetRudpPacket> packets,
long nowMs)
{
lock (_pendingReliableGate)
{
var available = Math.Max(0, ReliableSendWindowPackets - _pendingReliable.Count);
var ready = new List<CultNetRudpPacket>(Math.Min(available, packets.Count));
var ready = new List<CultNetRudpPacket>();
foreach (var packet in packets)
{
if (ready.Count < available)
if (_queuedReliable.Count == 0 && WindowAdmits(packet.Sequence, packet.Payload.Length))
{
_pendingReliable[packet.Sequence] = new PendingReliablePacket
{
@ -1660,9 +1690,9 @@ namespace GameCult.Networking
{
lock (_pendingReliableGate)
{
var available = Math.Max(0, ReliableSendWindowPackets - _pendingReliable.Count);
var ready = new List<CultNetRudpPacket>(Math.Min(available, _queuedReliable.Count));
while (ready.Count < available && _queuedReliable.Count > 0)
var ready = new List<CultNetRudpPacket>();
while (_queuedReliable.Count > 0
&& WindowAdmits(_queuedReliable.Peek().Sequence, _queuedReliable.Peek().Payload.Length))
{
var packet = _queuedReliable.Dequeue();
_pendingReliable[packet.Sequence] = new PendingReliablePacket
@ -1696,17 +1726,36 @@ namespace GameCult.Networking
{
lock (_pendingReliableGate)
{
_pendingReliable.Remove(packet.Ack);
Acknowledge(packet.Ack);
for (var bit = 0; bit < 32; bit++)
{
if ((packet.AckMask & (1u << bit)) != 0 && packet.Ack > bit)
{
_pendingReliable.Remove(packet.Ack - (uint)bit - 1);
Acknowledge(packet.Ack - (uint)bit - 1);
}
}
// Acknowledged sizes stop counting once the lowest unacknowledged sequence passes them, and all
// of them stop when nothing is unacknowledged.
if (_pendingReliable.Count == 0)
{
_ackedAboveLowest.Clear();
}
else
{
var lowest = _pendingReliable.Keys.Min();
foreach (var sequence in _ackedAboveLowest.Keys.Where(sequence => sequence <= lowest).ToArray())
_ackedAboveLowest.Remove(sequence);
}
}
}
// Callers hold the pending gate.
private void Acknowledge(uint sequence)
{
if (_pendingReliable.Remove(sequence, out var pending))
_ackedAboveLowest[sequence] = pending.Packet.Payload.Length;
}
private static bool PacketAcknowledges(CultNetRudpPacket packet, uint sequence)
{
if (packet.Ack == sequence)

235
tests/GameCult.Networking.Tests/NetworkingTests.cs
View file

@ -1141,6 +1141,196 @@ namespace GameCult.Networking.Tests
Assert.That(delivered[0].Payload, Is.EqualTo(payload));
}
// The sender's flow window: a reliable packet goes on the wire only while its sequence is at most 1,023
// above the lowest unacked one and the payload above that sequence stays within 4 MiB.
private const int Mib = 1024 * 1024;
private static CultNetRudpSession FlowSession(uint initialSequence) =>
new CultNetRudpSession(new CultNetRudpSessionOptions
{
ConnectionId = 0x464c4f57,
InitialSequence = initialSequence,
ResendDelayMs = 25
});
// A sender and a receiver that have shaken hands: the receiver's watermark is seeded by the sender's
// Connect, and the sender's Connect is acknowledged, so nothing is pending on the sender.
private static (CultNetRudpSession Sender, CultNetRudpSession Receiver) ConnectedFlowPair(uint senderInitialSequence, uint receiverInitialSequence)
{
var sender = FlowSession(senderInitialSequence);
var receiver = FlowSession(receiverInitialSequence);
sender.Receive(receiver.AcceptConnect(sender.CreateConnect(0), 0), 0);
return (sender, receiver);
}
private static CultNetRudpSession ConnectedFlowSession(uint initialSequence) => ConnectedFlowPair(initialSequence, 900).Sender;
private static IReadOnlyList<CultNetRudpPacket> FlowSend(CultNetRudpSession session, int payloadBytes) =>
session.SendMany("state", new byte[payloadBytes], new CultNetRudpSendOptions { Reliable = true });
private static CultNetRudpPacket FlowAck(uint sequence) =>
new CultNetRudpPacket { PacketType = CultNetRudpPacketType.Ack, ConnectionId = 0x464c4f57, Ack = sequence, ChannelId = "control" };
private static CultNetRudpPacket LoseGAndFillTheSpan(CultNetRudpSession session)
{
var g = FlowSend(session, 1).Single();
for (uint offset = 1; offset <= 1023; offset++)
{
Assert.That(FlowSend(session, 1).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + offset }));
session.Receive(FlowAck(g.Sequence + offset), 1);
}
return g;
}
[Test]
public void RudpSession_ALostPacketHoldsTheSender1023SequencesAheadAndIsStillDelivered()
{
var (sender, receiver) = ConnectedFlowPair(1, 100);
var g = FlowSend(sender, 1).Single();
var admittedAfterG = 0;
for (var index = 0; index < 4200; index++)
{
foreach (var packet in FlowSend(sender, 1))
{
admittedAfterG++;
receiver.Receive(packet, 1);
sender.Receive(receiver.CreateAck(), 1);
}
}
Assert.That(admittedAfterG, Is.EqualTo(1023));
Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(4200 - 1023));
var retransmit = sender.DueResends(1000).Single(packet => packet.Sequence == g.Sequence);
var delivered = receiver.Receive(retransmit, 1000).Delivered;
Assert.That(delivered.Select(frame => frame.Sequence), Is.EqualTo(new[] { g.Sequence }));
var promoted = sender.Receive(receiver.CreateAckForReceived(g.Sequence), 1001).ReadyToSend;
Assert.That(promoted.First().Sequence, Is.EqualTo(g.Sequence + 1024));
}
[Test]
public void RudpSession_ADirectSendPastTheSpanIsRefusedWithoutConsumingASequence()
{
var sender = ConnectedFlowSession(1);
var g = LoseGAndFillTheSpan(sender);
Assert.Throws<InvalidOperationException>(() => sender.Send("state", new byte[1], new CultNetRudpSendOptions { Reliable = true }));
Assert.That(sender.QueuedReliablePacketCount, Is.Zero);
sender.Receive(FlowAck(g.Sequence), 2);
var next = sender.Send("state", new byte[1], new CultNetRudpSendOptions { Reliable = true });
Assert.That(next.Sequence, Is.EqualTo(g.Sequence + 1024));
}
[Test]
public void RudpSession_PayloadAboveTheLowestUnackedSequenceIsBoundedBy4MiB()
{
var sender = ConnectedFlowSession(1);
var g = FlowSend(sender, Mib).Single();
// The lowest unacked packet's own bytes do not count: four more MiB fit above it.
for (var index = 0; index < 4; index++)
Assert.That(FlowSend(sender, Mib), Has.Count.EqualTo(1));
Assert.That(FlowSend(sender, Mib), Is.Empty);
Assert.That(FlowSend(sender, Mib), Is.Empty);
Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(2));
// Acking g moves the floor up one packet: exactly one MiB more fits.
var promoted = sender.Receive(FlowAck(g.Sequence), 1).ReadyToSend;
Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 5 }));
Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(1));
}
[Test]
public void RudpSession_APacketLargerThanTheWindowGoesOutAloneAndWaitsBehindAnythingUnacked()
{
Assert.That(FlowSend(ConnectedFlowSession(1), 5 * Mib), Has.Count.EqualTo(1));
var sender = ConnectedFlowSession(1);
var g = FlowSend(sender, 1).Single();
Assert.That(FlowSend(sender, 5 * Mib), Is.Empty);
var promoted = sender.Receive(FlowAck(g.Sequence), 1).ReadyToSend;
Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 1 }));
}
[Test]
public void RudpSession_ASmallPacketDoesNotOvertakeAQueuedLargeOne()
{
var sender = ConnectedFlowSession(1);
var g = FlowSend(sender, 1).Single();
Assert.That(FlowSend(sender, 3 * Mib), Has.Count.EqualTo(1));
Assert.That(FlowSend(sender, 2 * Mib), Is.Empty);
// A direct send refuses instead of queueing behind it, and takes no sequence.
Assert.Throws<InvalidOperationException>(() => sender.Send("state", new byte[1], new CultNetRudpSendOptions { Reliable = true }));
Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(1));
// One byte would fit above g, but the 2 MiB packet is ahead of it.
Assert.That(FlowSend(sender, 1), Is.Empty);
var promoted = sender.Receive(FlowAck(g.Sequence), 1).ReadyToSend;
Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 2, g.Sequence + 3 }));
}
[Test]
public void RudpSession_AcknowledgedBytesAboveALostPacketStillCountAgainst4MiB()
{
var sender = ConnectedFlowSession(1);
var g = FlowSend(sender, 1).Single();
// g is lost. Each 1 MiB frame above it is delivered and acknowledged, and the receiver still holds it
// behind the gap, so it keeps counting.
for (uint offset = 1; offset <= 4; offset++)
{
Assert.That(FlowSend(sender, Mib).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + offset }));
sender.Receive(FlowAck(g.Sequence + offset), 1);
}
Assert.That(FlowSend(sender, Mib), Is.Empty);
Assert.That(sender.QueuedReliablePacketCount, Is.EqualTo(1));
Assert.Throws<InvalidOperationException>(() => sender.Send("state", new byte[Mib], new CultNetRudpSendOptions { Reliable = true }));
// g arrives: nothing is held behind a gap any more, and the queue drains.
var promoted = sender.Receive(FlowAck(g.Sequence), 2).ReadyToSend;
Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 5 }));
// The bytes above the old gap no longer count: another MiB fits.
Assert.That(FlowSend(sender, Mib).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 6 }));
}
[Test]
public void RudpSession_AcknowledgedBytesCountAgainstTheLowestUnackedSequenceAsItAdvances()
{
var sender = ConnectedFlowSession(1);
var g = FlowSend(sender, 1).Single();
var h = FlowSend(sender, 1).Single();
for (var index = 0; index < 3; index++)
FlowSend(sender, Mib);
// g and the 3 MiB above h are acknowledged; h is lost.
foreach (uint offset in new uint[] { 2, 3, 4, 0 })
sender.Receive(FlowAck(g.Sequence + offset), 1);
Assert.That(FlowSend(sender, Mib).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 5 }));
// 4 MiB sit above h, acknowledged or not: one more byte does not fit.
Assert.That(FlowSend(sender, 1), Is.Empty);
// h arrives, the floor moves to g + 5, and the byte fits.
var promoted = sender.Receive(FlowAck(h.Sequence), 2).ReadyToSend;
Assert.That(promoted.Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + 6 }));
}
// A server that accepts a fresh Connect may send before its Accept is acknowledged, so no acknowledgement of
// the new generation has moved the floor yet. Bytes the ended generation had acknowledged above its own lost
// packet belong to no receiver's hold any more and must not count.
[Test]
public void RudpSession_ANewGenerationOnTheServerDoesNotInheritAcknowledgedBytes()
{
var server = FlowSession(500);
var accept = server.AcceptConnect(FlowSession(1).CreateConnect(0), 0);
server.Receive(FlowAck(accept.Sequence), 0);
var g = FlowSend(server, 1).Single();
for (uint offset = 1; offset <= 4; offset++)
{
Assert.That(FlowSend(server, Mib).Select(packet => packet.Sequence), Is.EqualTo(new[] { g.Sequence + offset }));
server.Receive(FlowAck(g.Sequence + offset), 1);
}
Assert.That(FlowSend(server, Mib), Is.Empty);
var nextAccept = server.AcceptConnect(FlowSession(200_000).CreateConnect(2), 2);
Assert.That(nextAccept.PacketType, Is.EqualTo(CultNetRudpPacketType.Accept));
// The Accept is unacknowledged and the only thing pending: 1 MiB fits above it.
var sent = server.Send("state", new byte[Mib], new CultNetRudpSendOptions { Reliable = true });
Assert.That(sent.Sequence, Is.EqualTo(nextAccept.Sequence + 1));
}
[Test]
public void RudpSocketTransport_HandshakesAndCarriesReliableOrderedSchemaFrames()
{
@ -5262,10 +5452,49 @@ namespace GameCult.Networking.Tests
#pragma warning restore CS0618
}
[Test]
public void CultMeshRudpClient_DefaultBindAllowsRemoteRoutes()
private static string? RudpClientBoundHost(string endpoint, string? bindHost = null)
{
Assert.That(new CultMeshRudpSocketOptions().BindHost, Is.EqualTo("0.0.0.0"));
using var client = CultMesh.CreateRudpClient(
"csharp-cultmesh-bind-client",
0x10203060u,
endpoint,
new CultMeshRudpSocketOptions { BindHost = bindHost });
return client.Profile.Transports[0].Host;
}
// A client bound to loopback cannot send to another host on Windows, so a client for a
// remote endpoint binds the unspecified address of its family.
[Test]
public void CultMeshRudpClient_ForARemoteEndpoint_BindsTheUnspecifiedAddress()
{
Assert.That(RudpClientBoundHost("rudp://10.77.0.1:17872"), Is.EqualTo("0.0.0.0"));
Assert.That(RudpClientBoundHost("rudp://[2001:db8::1]:17872"), Is.EqualTo("::"));
}
[Test]
public void CultMeshRudpClient_ForALoopbackEndpoint_BindsLoopback()
{
Assert.That(RudpClientBoundHost("rudp://127.0.0.1:17872"), Is.EqualTo("127.0.0.1"));
Assert.That(RudpClientBoundHost("rudp://[::1]:17872"), Is.EqualTo("::1"));
}
[Test]
public void CultMeshRudpClient_BindsAnExplicitBindHostWhateverTheEndpoint()
{
Assert.That(RudpClientBoundHost("rudp://10.77.0.1:17872", "127.0.0.1"), Is.EqualTo("127.0.0.1"));
Assert.That(RudpClientBoundHost("rudp://127.0.0.1:17872", "0.0.0.0"), Is.EqualTo("0.0.0.0"));
}
[Test]
public void CultMeshRudpServer_BindsEveryIPv4InterfaceUnlessToldOtherwise()
{
using var anywhere = CultMesh.CreateRudpServer("csharp-cultmesh-bind-server", 0x10203061u);
using var loopback = CultMesh.CreateRudpServer(
"csharp-cultmesh-bind-server",
0x10203062u,
new CultMeshRudpSocketOptions { BindHost = "127.0.0.1" });
Assert.That(anywhere.Profile.Transports[0].Host, Is.EqualTo("0.0.0.0"));
Assert.That(loopback.Profile.Transports[0].Host, Is.EqualTo("127.0.0.1"));
}
[Test]