Compare commits

...
Sign in to create a new pull request.
Author SHA1 Message Date
7665df33ae Merge pull request 'TS: write_admitted / error.v1 decoder, WriteTracker, vector replay runner' (#62) from libby/ts-write-receipt-client into main
Some checks failed
CultMesh QUIC native / Build win32-x64 (push) Has been skipped
CultMesh QUIC native / Build linux-x64 (push) Has been skipped
CultMesh QUIC native / Release native binaries (push) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (push) Failing after 5m10s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (push) Successful in 31s
2026-11-08 19:52:37 +00:00
92b4cb7ce9 Merge pull request 'C#: database server answers writes with write_admitted / error.v1 (write-receipt spec)' (#61) from libby/cs-write-receipt-server into main
Some checks failed
CultMesh QUIC native / Build win32-x64 (push) Has been skipped
CultMesh QUIC native / Build linux-x64 (push) Has been skipped
CultMesh QUIC native / Release native binaries (push) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (push) Has been cancelled
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (push) Has been cancelled
2026-11-08 19:51:52 +00:00
36494dafb9 Merge pull request 'cultmesh-rs: RUDP document server answers writes with write_admitted / error.v1' (#60) from libby/rs-write-receipt-server into main
Some checks failed
CultMesh QUIC native / Build win32-x64 (push) Has been skipped
CultMesh QUIC native / Build linux-x64 (push) Has been skipped
CultMesh QUIC native / Release native binaries (push) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (push) Has been cancelled
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (push) Has been cancelled
2026-11-08 19:49:50 +00:00
5bf2428295 Merge pull request 'spec: cultnet write receipt (write_admitted, error id echo, admitted/refused/unconfirmed)' (#59) from libby/cultnet-write-receipt-spec into main
Some checks failed
CultMesh QUIC native / Build win32-x64 (push) Has been skipped
CultMesh QUIC native / Build linux-x64 (push) Has been skipped
CultMesh QUIC native / Release native binaries (push) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (push) Has been cancelled
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (push) Has been cancelled
2026-11-08 19:49:40 +00:00
Libby
621b86ca15 Spec #59: an empty or blank schemaId or recordKey is message_invalid (details.field message, the write's id), never applied; vector for it
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 5m55s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 46s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 19:45:19 +00:00
Libby
90d252735f cultmesh-rs: a refused reliable-sequenced write is received and acknowledged first, and a refused write counts as session activity
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 6m11s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 52s
Refusing before session.receive left the packet unacknowledged: the client resent it forever and every later ordered write stalled. The test now uses the same peer and proves a later ordered write is admitted.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 19:38:43 +00:00
Libby
dc1ae51e92 cultmesh-ts: publishRudpDocumentOnce builds the put before connecting, so a put that cannot be built keeps its cause on both paths
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 6m13s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 31s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 19:37:08 +00:00
Libby
0fbf2baea6 cultnet-ts, cultmesh-ts: Heimdall's five Lows on send-only (spec wording, error.v1 ack-time refusal tests, build errors keep their cause, presence changelog, ack-cut anchors)
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Has been cancelled
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Has been cancelled
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 19:32:49 +00:00
Libby
89ff3d2cbd cultnet-ts test: the Odin presence publisher is send-only, so a refusal after the transport ack is not awaited
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Has been cancelled
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Has been cancelled
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 19:26:25 +00:00
Libby
206dbe142b cultmesh-ts: sendOnly tests use one fake server per publish (the reused one threw on the second connect and hung the run); spec names send-only and the presence publisher
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Has been cancelled
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Has been cancelled
The hang was the fake server, not the writer: it is bound to one connection id, and two publishes in one test sent a second connect it rejected from inside the socket handler. The unacknowledged-send test also scripted an admit, whose reply piggybacks the put's ack.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 19:24:52 +00:00
Libby
0caef4d564 WIP (unverified): publishRudpDocumentOnce sendOnly + presence publisher send-only; tests written, cultmesh-ts receipt-client run hung
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 19:24:52 +00:00
Libby
c89c4ee39a cultmesh-ts: test kills the flush-cap mutant
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 5m58s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 1m18s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 17:07:54 +00:00
Libby
af1ac72359 cultmesh-ts, cultnet-ts: only write_admitted admits a put (no echo admission); the send is bounded by the result window and a timeout is local_send_failed
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 17:05:42 +00:00
Libby
293b4e83bc cultmesh-ts, cultnet-ts: an unanswered put is unconfirmed on every server, session_ended before a result is an error, a failed local send is its own reason
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 17:03:39 +00:00
Libby
a67faad27c cultnet-ts: tests for two tracker mutants (send expires other windows; non-answer frames settle nothing)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 17:03:39 +00:00
Libby
6eea66c57f cultnet-ts, cultmesh-ts: write receipt client (frames, WriteTracker, replay runner, opt-in writer)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 17:03:39 +00:00
Libby
e4b67a9e0b cultmesh-rs: a receipt session accepts writes only on a reliable, non-sequenced channel; a test pins that a reconnect starts a fresh session
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 10m5s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 1m8s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 17:00:48 +00:00
Libby
74adad916a Spec #59: receipt writes need a reliable, non-sequenced channel; error.v1 schema description names limit_reached; #55-#58 paragraph states which clients still window-succeed
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 9m8s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 1m14s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 16:59:39 +00:00
Libby
1a05df2524 cultmesh-rs: server-wide remembered-write budget is a maintained counter; a session whose refusal set overflowed refuses unseen ids after the budget frees
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 5m35s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 1m1s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 16:36:10 +00:00
Libby
2c5203dc82 Spec #59: local_send_failed is its own unconfirmed reason
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 5m31s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 31s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 16:34:52 +00:00
Libby
3443e1dc93 Spec #59: error.v1 code enum names limit_reached; server-wide budget is a conformance column; an overflowed refusal set refuses unseen ids for the session
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Has been cancelled
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Has been cancelled
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 16:34:19 +00:00
Libby
c8f869e54e cultnet-rs, cultmesh-rs: null or missing document/schemaId refused naming the field; budget tests use literal bytes; unreliable-refusal local failure test
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 5m33s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 26s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 15:32:03 +00:00
Libby
d7b36416c6 cultmesh-rs: pin limit-refusal memory bound and legacy unreliable drop (mutant-driven tests)
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Has been cancelled
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Has been cancelled
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 15:30:36 +00:00
Libby
1447ccd8dd cultnet-rs, cultmesh-rs: write_admitted in spec field order; server-wide remembered-write budget over digest entries; id size limit; limit refusals remembered; unreliable writes refused; local failure is its own goodbye
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 5m14s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 46s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 15:09:28 +00:00
Libby
734087766a Spec #59: id size, digest memory, limit refusals remembered, required fields, reliable channel, local send failure, per-class conformance table
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 6m0s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 46s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 15:04:21 +00:00
Libby
b89a883700 spec: C# RUDP schema server is not receipt-capable; advertise only if end to end
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Has been cancelled
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Has been cancelled
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 14:58:51 +00:00
Libby
0182a6d40b cultnet-rs, cultmesh-rs: remembered-write cap refusal is error.v1 limit_reached; pin authorize-once with a counting sink
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 5m19s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 41s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 14:32:49 +00:00
Libby
eafcdcdf7c cultnet: add limit_reached to error.v1 code enum, with captured vector
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 4m51s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 26s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 14:29:18 +00:00
Libby
5f699c27fa cultmesh-rs: a put reusing a delete's id is refused (mutant-driven test)
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Has been cancelled
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Has been cancelled
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 14:27:54 +00:00
Libby
d4f440ac7c cultmesh-rs: receipt server looks the id up first, remembers refusals, answers deletes and unqueueable refusals
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 4m47s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Has been cancelled
Follows Heimdall's second review of #60: replay by (session,id) before any
authorization; refusals stored with their hash; delete shares the dedupe
table and settles capability; a refusal the queue cannot take ends the
session; a peer's refusal frames and a receipt session's refused deletes are
not violations; the write limit has its own fixed text; the sink trait says
Ok means committed and visible.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 14:22:50 +00:00
Libby
7a560b1984 spec: dedupe scope is at least the session; a resend after a reconnect is a new write
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 5m52s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 41s
Per Heimdall's review. The dedupe key lives for the life of its session, bounded per
session without forgetting an id the server promised to answer. A longer retention
is optional and never to be relied on. After a reconnect callers resend only writes
that are idempotent by record key and read back before resending any other.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 13:58:39 +00:00
Libby
8bd31a5d4a cultmesh-rs: receipt server follows Heimdall's review of #60
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 7m11s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 26s
A refused put, a full dedupe table and a reused id refuse the write, not the
session; only a protocol violation ends one. A frame that does not decode but
holds a messageId is refused with that id. The server answers a receipt-capable
hello with its own, listing both frames. The sink's authorization (new
authorize_raw_document, default admit) runs before the session's remembered
writes are consulted. A second hello does not change the capability: it is set
once, by the first hello or the first write. A receipt that cannot be queued, and
a write the payload budget would drop silently, end the session instead of
leaving the write unanswered. Refusals to a receipt-capable session are queued
reliably.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 13:56:17 +00:00
Libby
9f73047c29 cultmesh-rs: RUDP document server answers writes with write_admitted / error.v1 (write-receipt spec)
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Has been cancelled
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Has been cancelled
cultnet-rs gains cultnet.write_admitted.v0, cultnet.error.v1 (v0 + messageId) and
write_content_hash. The server opts a session in only when its hello lists both frames
(framing only; admission rules are unchanged). On an opted-in session an admitted put is
answered with a receipt carrying its id after the sink accepts it, a refusal is an error.v1
naming the id, a resend of the same content is re-receipted without reaching the sink, and the
same id with other content is refused message_invalid (never the old receipt). Other sessions
are answered exactly as before.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 13:38:46 +00:00
Libby
ab7bc771e3 spec: write receipt per Heimdall's review (dedupe by actor+content hash, admitted-late, 5 s window, capability is framing only)
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 6m28s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 2m19s
- messageId unique per authenticated actor and session; dedupe key (actor, id) bound to SHA-256 of
  [operation, schemaId, recordKey, payload]; same id with other content is message_invalid, never the old receipt.
- A late answer after the window is recorded on the write (admitted_late / refused_late); no blind resend.
- A forwarder with no primary answer relays nothing; the client reports unconfirmed.
- RECEIPT_RESULT_WINDOW_MS 5000 (default), LEGACY_RESULT_WINDOW_MS 500 (fallback only), DEDUPE_RETENTION_MS as vector constants.
- The hello capability changes framing only and never authorizes.
- Batches can commit partially; the caller owns retry.
- Vectors: constants, worked content hashes, 7 dedupe cases, late-answer and window scenarios.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 13:31:49 +00:00
Libby
662ae048d8 spec: cultnet write receipt (write_admitted.v0, error.v1 id echo, three-outcome clients)
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Has been cancelled
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Has been cancelled
Specification and shared vectors only, no runtime code. A write is answered
exactly once with cultnet.write_admitted.v0 or cultnet.error.v1 carrying its
messageId; the new frames go only to connections whose hello lists them (today's
TS decoder rejects unknown keys and versions, so "old clients ignore it" is
false). Clients settle admitted | refused | unconfirmed; the result window is the
pre-receipt fallback and ends unconfirmed. Wire vectors, schema-invalid cases and
16 replay scenarios in write-receipt-vectors.json.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-11-08 13:28:46 +00:00
26 changed files with 4448 additions and 184 deletions

View file

@ -115,6 +115,7 @@ copies), the C# `CultNetErrorMessage` factory, the Rust `CultNetErrorCode`, and
`unowned_schema` (a write, or a remote put or delete, for a schema and key no shard owns), `unowned_schema` (a write, or a remote put or delete, for a schema and key no shard owns),
`write_not_authorized` (a peer's put or delete its server's remote write authority did not grant) and `write_not_authorized` (a peer's put or delete its server's remote write authority did not grant) and
`unsupported_schema_version` (a message whose `schemaVersion` the peer does not speak; `details` names `schemaVersion` and the offered value) and `unsupported_schema_version` (a message whose `schemaVersion` the peer does not speak; `details` names `schemaVersion` and the offered value) and
`limit_reached` (a receipt-capable server's per-session remembered-write table is full, so a new write is refused; `details` all null) and
`message_invalid` (a request that lacks a required field or carries an ill-typed one: a document-put or delete, or a subscription without an id; `details.field` names the request field, or `message` when the whole frame cannot be read). `message_invalid` (a request that lacks a required field or carries an ill-typed one: a document-put or delete, or a subscription without an id; `details.field` names the request field, or `message` when the whole frame cannot be read).
A malformed put or delete is `message_invalid`, not a per-operation code: the peer knows which operation it sent. A malformed put or delete is `message_invalid`, not a per-operation code: the peer knows which operation it sent.

View file

@ -0,0 +1,122 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://github.com/GameCult/CultLib/contracts/cultnet/interop/cultnet.error.v1.schema.json",
"title": "CultNet Error Message v1",
"description": "cultnet.error.v0 plus messageId (write-receipt.md). Sent in place of cultnet.error.v0 only to a connection that declared the write-receipt capability. v0 is unchanged: its schema is closed (additionalProperties false), so an extra key would make a v0 peer reject the frame instead of ignoring it. Map-mode: every key is written, messageId nil when the error answers no readable message. The code enum is the v0 enum plus limit_reached (write-receipt.md, Remembered-write limits).",
"type": "object",
"required": [
"schemaVersion",
"error",
"routingHint",
"code",
"details",
"messageId"
],
"additionalProperties": false,
"properties": {
"schemaVersion": {
"const": "cultnet.error.v1"
},
"error": {
"type": "string",
"minLength": 1
},
"routingHint": {
"description": "Optional shard routing detail for an authority failure. Null for every selection/cursor refusal code (R-N).",
"type": [
"object",
"null"
],
"additionalProperties": false,
"required": [
"shard",
"reason"
],
"properties": {
"shard": {
"type": [
"object",
"null"
]
},
"reason": {
"type": "string"
}
}
},
"code": {
"description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it. The set is a closed enum: Rust and TypeScript refuse a message whose code is not listed, so adding a code is one coordinated change across every runtime (this schema, the C# factory, the Rust CultNetErrorCode, and a captured-bytes case for each). limit_reached: a receipt-capable server refuses a NEW write because its per-session remembered-write table (1024 entries) or its server-wide memory budget is full; messageId is the write's id.",
"type": [
"string",
"null"
],
"enum": [
null,
"selection_invalid",
"cursor_stale",
"cursor_invalid",
"reference_outside_target",
"variant_unsupported",
"unowned_schema",
"write_not_authorized",
"unsupported_schema_version",
"message_invalid",
"limit_reached"
]
},
"details": {
"description": "Code-specific structured detail (R-N). Every field is populated only for the code that defines it; the others are null, not omitted.",
"type": [
"object",
"null"
],
"additionalProperties": false,
"required": [
"field",
"value",
"asOf",
"current"
],
"properties": {
"field": {
"description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended. unowned_schema: \"schemaId\". write_not_authorized: \"recordKey\". unsupported_schema_version: \"schemaVersion\". message_invalid: the request field that is missing or ill-typed (\"message\" when the whole frame cannot be read). limit_reached: no field; details stay all null. message_invalid also names \"messageId\" (over 256 bytes), \"document\", \"schemaId\" or \"channel\" (write on an unreliable channel).",
"type": [
"string",
"null"
]
},
"value": {
"description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended. unowned_schema: the schema id no shard owns. write_not_authorized: the key of the refused write. unsupported_schema_version: the schemaVersion the peer does not speak.",
"type": [
"string",
"null"
]
},
"asOf": {
"description": "cursor_stale: the cursor's own minted asOf.",
"type": [
"integer",
"null"
],
"minimum": 0
},
"current": {
"description": "cursor_stale: the answering server's current asOf.",
"type": [
"integer",
"null"
],
"minimum": 0
}
}
},
"messageId": {
"description": "The messageId of the write (or other request) this error answers, echoed byte for byte. Null only when the frame was unreadable or carried no messageId.",
"type": [
"string",
"null"
],
"minLength": 1
}
}
}

View file

@ -0,0 +1,16 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://github.com/GameCult/CultLib/contracts/cultnet/interop/cultnet.write-admitted.schema.json",
"title": "CultNet Write Admitted",
"description": "The receipt for one admitted document put or delete (write-receipt.md). Sent only after the write's commit point, only to a connection that declared the write-receipt capability, once per messageId. Map-mode: every key is written, none is nil. A receipt never carries an error, and an error.v1 never carries admission.",
"type": "object",
"required": ["schemaVersion", "messageId", "operation", "schemaId", "recordKey"],
"additionalProperties": false,
"properties": {
"schemaVersion": { "const": "cultnet.write_admitted.v0" },
"messageId": { "description": "The messageId of the put or delete this admits, echoed byte for byte.", "type": "string", "minLength": 1 },
"operation": { "enum": ["put", "delete"] },
"schemaId": { "description": "The schemaId of the admitted write, echoed.", "type": "string", "minLength": 1 },
"recordKey": { "description": "The recordKey of the admitted write, echoed.", "type": "string", "minLength": 1 }
}
}

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,83 @@
# CultNet write receipt (specification)
Status: specification. No runtime implements it yet; the conformance follow-ups below do.
Rules: Heimdall's ruling of 2026-11-08 (a write completes on its result frame; an ack or a return is not admission).
Files here: `cultnet.write-admitted.schema.json`, `cultnet.error.v1.schema.json`, `write-receipt-vectors.json`. They move to `contracts/cultnet/` when the first runtime implements them.
## The problem
A put or delete today has two answers on the wire: nothing (admitted, or the server never read it) and `cultnet.error.v0` (refused, with no id saying which write). A transport ack proves delivery. Clients that completed on the ack, or on a window with no refusal, reported refused writes as done (CultLib #55 to #58 bound this with a result window; this spec replaces the window with an answer).
## Frames
**`cultnet.write_admitted.v0`**: `{ schemaVersion, messageId, operation: "put"|"delete", schemaId, recordKey }`, every key written, none nil. Sent once a put or delete is admitted. The id echoes the write's `messageId` byte for byte; `operation`, `schemaId`, `recordKey` echo the write.
**`cultnet.error.v1`**: `cultnet.error.v0` plus a required `messageId` (string or nil). The code enum is v0's plus `limit_reached` (see Remembered-write limits); `cultnet.error.v1.schema.json` is the authority for the closed set. Sent in place of `error.v0` to a receipt-capable connection. `messageId` is the id of the request the error answers; nil only when the frame was unreadable or carried no id.
Why a new error version and not an extra key on v0: the v0 schema is closed (`additionalProperties: false`) and the TypeScript decoder (Ajv over the closed schemas) rejects unknown keys and unknown `schemaVersion`s, and `parseCultNetMessage` throws on an unknown version; Rust's decoder errors on an unknown version too. Unknown-key tolerance of the Rust, Python and Kotlin error decoders was not checked, so none is assumed, so a surprise key or frame breaks an old client's whole receive path. Old clients are therefore never sent either new frame. This reverses the "old clients ignore it" assumption, which is false for today's decoders.
## Capability: opting in
A connection is receipt-capable when its `cultnet.hello.v0` (sent before its first write; the schema channel is reliable and ordered) lists both `cultnet.write_admitted.v0` and `cultnet.error.v1` in `supportedMessageVersions`. A server MUST NOT send either new frame to any other connection, and answers it exactly as before (no receipt, `error.v0`). A receipt-capable server lists both in its own hello, so a client can tell a legacy server from a receipt server that has not answered yet. A client that sends no hello gets legacy behaviour. A server advertises the two frames in its hello only if it implements the receipt rules end to end.
**The capability changes framing only; it never authorizes.** A hello authenticates nothing and grants nothing: who may write, and which keys, is decided by the server from the transport-authenticated actor exactly as without the capability. Declaring the capability never makes a refused write admitted, never widens an actor's authority, and never changes what a write does; it selects only which frame the answer arrives in.
## Server rules
1. **One answer.** For every put or delete on a receipt-capable connection the server sends exactly one of `write_admitted` or `error.v1`, carrying that write's `messageId`. Never both, never two different answers.
2. **After the commit point.** `write_admitted` is sent only after the write is committed by the same rule that makes it visible to readers. Receipt of the bytes, parse, or authorisation alone is not admission.
3. **Id on every refusal.** Every refusal of a write, whatever its code (`write_not_authorized`, `message_invalid`, `unowned_schema`, `unsupported_schema_version`, shard authority errors, ...), is an `error.v1` whose `messageId` is the write's id. When the frame is readable enough to find the id, it is found; `message_invalid` for a missing or ill-typed field still carries the id. Nil only when no id can be read.
4. **Batches.** There is no batch frame; a batch is N writes, each with its own `messageId`, each answered under rules 1 to 3. A failed batch names its member by that id. A server may not collapse a batch into one answer.
5. **Identity and idempotent resend.** A `messageId` is unique per authenticated actor and session: the client mints it, never reuses it for different content in that session, and a server never accepts the id of one actor as the id of another. The server's dedupe key is `(actor, messageId)`, where *actor* is the transport-authenticated identity, never a name the hello or the frame claims. Alongside the key it keeps the **content hash** of the write and the answer it gave. The content hash is SHA-256 of the MessagePack array `[operation, schemaId, recordKey, payload]`, where `payload` is the raw document payload as a MessagePack `bin` for a put and nil for a delete (the vectors give worked hashes). **Scope: at least the session.** The key lives for the life of the session it was made in, and a server may bound how many writes it remembers per session (see Remembered-write limits; a server never forgets an id it promised to answer). A server MAY keep the key longer for the same actor, up to `DEDUPE_RETENTION_MS` (10 minutes), but a client MUST NOT rely on it: **a reconnect is a new session, and a resend on the new session is a new write.**
- Same key, same content hash, answer was `write_admitted`: answer `write_admitted` again with the same body. The write is not applied twice.
- Same key, same content hash, answer was a refusal: the same refusal again.
- **Same key, different content hash: refuse `message_invalid` (details `field: "messageId"`) and never re-send the old receipt.** Old receipts are for the old content; showing one for new content would report an unapplied write as admitted. The original write's state is unchanged.
- **Id size.** A write's `messageId` is at most 256 bytes of UTF-8. A longer id is refused `message_invalid` (`details.field: "messageId"`, the refusal's `messageId` the id as received, text fixed per Fixed texts) BEFORE anything is remembered: no table entry, no digest, no budget charge. Such a refusal is therefore not replayed from the table; a resend is refused afresh, identically.
- **What is remembered.** Per entry a server keeps only fixed-size digests (SHA-256) of the `messageId` and of the content hash, plus the answer (`admitted`, or the refusal's code and details). Memory per entry is bounded independent of id length and payload size. Lookup compares digests.
- **Remembered-write limits.** The per-session table holds at most 1024 entries. A server MAY also enforce a server-wide memory budget across all sessions (reference: 32 MiB of remembered entries). A NEW write (a key not already in the table) that would exceed either bound is refused `limit_reached` for that write alone, `messageId` the write's id, `details` all null; the write is not applied and the session continues. A resend of a remembered key is still answered from the table.
- **Limit refusals are remembered.** A `limit_reached` refusal is remembered like any other: a resend of the same id and content hash replays `limit_reached`. No slot is freed mid-session, so an id refused at a bound never becomes admitted later in that session (a server at the cap keeps the refused digest pair in a bounded refusal set of at most 1024 per session, outside the entry budget; past it, a further new write is refused `limit_reached` without being remembered, and the session refuses every id it has not seen for the rest of its life, even if a server-wide budget has since freed, so a refusal that was not remembered is never followed by an admission of the same id). The client mints a new id on a new session.
- **Required fields.** A put or delete whose document or `schemaId` is null or missing is refused `message_invalid` with `details.field` `"document"` or `"schemaId"` (the first missing in that order) and fixed text; the write is not applied. A `schemaId` or `recordKey` that is empty or only whitespace (present but names nothing) is not "missing": the write is refused `message_invalid` exactly as a frame that does not read, with `details.field` `"message"` and the write's id (vector `error_v1_empty_schema_id_is_unreadable_and_names_the_write`), and is never applied; a server must not hand it to its store.
- **Fixed texts.** The `error` string of these refusals is fixed, so vectors can compare it: `message_invalid: messageId exceeds 256 bytes`, `message_invalid: document is required`, `message_invalid: schemaId is required`, `message_invalid: receipt sessions require the reliable channel`, `limit_reached: the session's remembered-write limit is full`.
- **Reliable channel only.** A write is accepted on a receipt session only on a channel that is reliable and not sequenced (reliable ordered and reliable unordered both qualify; a sequenced channel drops superseded packets, so a write on it can vanish without trace). A write arriving on any other channel is refused `message_invalid` with `details.field: "channel"` (the transport profile's word for it) and fixed text. Receipts and refusals travel the reliable channel; a write that could be lost without trace cannot be answered once.
- Past the session (or a server's longer retention, if it keeps one) a resend is a new write.
6. **Forwarded writes.** A server that forwards a non-primary write answers the original peer only with what the primary answered: the primary's receipt becomes `write_admitted` with the original id, its refusal becomes `error.v1` with the original id. If the forward gets no answer the server relays nothing: no receipt, no error, no synthesized refusal. The client's own rule below then reports the write as unconfirmed. A forwarder never turns an absent or ambiguous primary answer into an admission.
7. **Looks alike.** An admitted and a refused write differ in `schemaVersion`; a receipt has no `error`, an `error.v1` has no admission. A transport ack is neither.
## Client rules: three outcomes
A write settles as exactly one of:
- **admitted**: a `write_admitted` whose `messageId` is the write's.
- **refused**: an `error.v1` whose `messageId` is the write's, carrying its `code` and `details` as the typed error.
- **unconfirmed**: nothing settled it. Reasons: `window_elapsed` (the result window ran from the *send* and passed with no answer; see Result window), `session_ended` (the peer closed the session, or it ended, before the write was answered) and `local_send_failed` (the client's own send of this write failed). A local send failure is never reported as refused: no server answer exists, and the write may or may not have left the machine. Unconfirmed is not success and not refusal. APIs that return a value return the tri-state; APIs that throw throw a distinct refused error and a distinct unconfirmed error and return normally only for admitted. A caller that needs admission treats unconfirmed as not admitted. Resend with the same `messageId` and unchanged content, **on the same session**, is safe against a receipt-capable server (rule 5). After a reconnect the resend is a new write, and a legacy server may apply a resend twice: the caller resends only writes that are idempotent by record key (a put of the same content to the same key, a delete of a key), and reads the record back before resending any other.
**Result window.** Two named constants (also in the vectors): `RECEIPT_RESULT_WINDOW_MS = 5000`, the default against a receipt-capable server (its hello lists both frames); `LEGACY_RESULT_WINDOW_MS = 500`, used only as the fallback against a server whose hello does not list them (or that sent none). The window runs from the send. A receipt-capable server that has not answered within its window is reported as unconfirmed like any other; clients log it as a server that missed its window.
**Late answers.** An answer for a write that already settled `unconfirmed` does not change what the caller was told, but the client records it on the write: a late `write_admitted` makes the write's recorded state `admitted_late`, a late refusal makes it `refused_late`. The write handle exposes that state, and a caller that got `unconfirmed` MUST consult it (and the record) before resending: it must not blind-resend a write that was admitted late. A resend on the same session with the same id and content is harmless on a receipt-capable server (rule 5); across a reconnect the caller follows the idempotence rule above.
**Legacy answers.** An `error.v0`, or an `error.v1` with nil `messageId`, names no write. It is attributed to a write only when exactly one write on the connection is unanswered; it then refuses that write. With two or more unanswered writes it settles none of them and each ends unconfirmed at its window. A frame naming a `messageId` that is not in flight settles nothing.
**Batches can commit partially.** A batch is N independent writes with no atomicity: some members may be admitted while others are refused or unanswered, and nothing is rolled back. The client reports each member's outcome and a batch result is only their set: admitted if every member is admitted, refused if any member is refused (naming the members), otherwise unconfirmed (naming the unanswered members). The caller owns retry: it resends exactly the members that are not admitted, with their original ids.
## Changes to what #55 to #58 shipped
Those clients complete *successfully* when the window ends with no refusal. Under this spec that case is `unconfirmed`. The TypeScript one-shot writer (#62) already reports it so: it resolves `admitted` only on a `write_admitted` carrying the put's id, and no other frame, a put echoed under the same id included, ever counts as admission. A caller whose server cannot answer a write yet (Odin, until it is receipt-capable) opts into the explicit send-only mode per call (`sendOnly: true`), which resolves the distinct outcome `sent` once the transport has acknowledged the put and never `admitted`. The Idunn presence publisher is send-only by its own function (`sendPresenceDocument`, not the `sendOnly` option): the same `sent` result, plus a rejection if a refusal naming the put (or naming nothing) arrives by the transport ack. Receipts and send-only are both opt-in, neither is the default. The C# and Rust clients of #55 to #58 keep the window-success until their conformance work lands, and callers of those must not read it as admission.
## Vectors
`write-receipt-vectors.json`: `constants` (the named windows and retention), `contentHash` (worked dedupe hashes), `dedupe` (server replay cases for rule 5), `wire` (MessagePack hex plus the decoded message for each new frame; the bytes are spec-authored with msgpack-python and every runtime's first conformance PR re-captures them with its own encoder and must match), `schemaInvalid` (frames that must be refused by a validator, each with why), and `scenarios` (event replays with the outcome each write must settle as; the clock is abstract; each case names its window constant). A conformant client passes every scenario; a conformant server's answers for the same inputs satisfy server rules 1 to 7.
## Conformance status
| Server class | Advertises both frames | Server-wide memory budget (MAY, Remembered-write limits) |
|---|---|---|
| `CultNetDatabaseServer` | yes | not implemented: only the per-session limit of 1024 applies |
| Shard write forwarder | yes | not implemented: only the per-session limit of 1024 applies |
| Rust RUDP document server | yes | implemented (32 MiB default, `max_remembered_write_bytes`) |
| C# RUDP schema server (`CultNetRudpSchemaServer`, `src/GameCult.Networking/CultNetRudpSchemaServer.cs`) | no | n/a |
Advertising is per server class, not per transport. A class advertises both frames in its hello only if it implements the receipt rules end to end; the C# RUDP schema server stays "no" until it answers writes with receipts.
## Conformance follow-ups (not in this PR)
C# server and client, Rust, TypeScript, Python, Kotlin: emit and parse the two frames; hello capability; tri-state client outcome; replace the #55 to #58 window-success; re-capture the wire vectors in each runtime. Server write paths include the shard write forwarder (rule 6) and the Odin catalog.

View file

@ -441,8 +441,8 @@ consumer that waits wants delivered, which is now just Acknowledged.
| `cultmesh-rs/src/lib.rs:781-838` `request_raw_snapshot_from_rudp_catalog` (Idunn `src/drivers.rs:5527` uses it) | a hand-rolled loop until `Acknowledged`, then the response | Cut 4: `wait_acknowledged` | | `cultmesh-rs/src/lib.rs:781-838` `request_raw_snapshot_from_rudp_catalog` (Idunn `src/drivers.rs:5527` uses it) | a hand-rolled loop until `Acknowledged`, then the response | Cut 4: `wait_acknowledged` |
| `cultmesh-rs/src/lib.rs:870-900` `publish_cultnet_message_to_rudp_catalog` (Odin `odin-daemon`) | a hand-rolled loop until `Acknowledged`, then disconnect. The doc at `:869` claims "application admission" | Cut 4: `wait_acknowledged`; the doc says delivered to the document server's sink (`rudp_document_server.rs:343-372`) | | `cultmesh-rs/src/lib.rs:870-900` `publish_cultnet_message_to_rudp_catalog` (Odin `odin-daemon`) | a hand-rolled loop until `Acknowledged`, then disconnect. The doc at `:869` claims "application admission" | Cut 4: `wait_acknowledged`; the doc says delivered to the document server's sink (`rudp_document_server.rs:343-372`) |
| `cultmesh-rs/src/lib.rs:918-938` `publish_cultnet_messages_to_rudp_catalog` (Ghostlight `mesh.rs:323`) | `flush_reliable`, then drop. **Several ordered frames, then close** | None; flush now means it | | `cultmesh-rs/src/lib.rs:918-938` `publish_cultnet_messages_to_rudp_catalog` (Ghostlight `mesh.rs:323`) | `flush_reliable`, then drop. **Several ordered frames, then close** | None; flush now means it |
| `cultmesh-ts/src/index.ts:6070-6098` `publishRudpDocumentOnce` (Heimdall, StreamPixels, Stonks, weksa, Bifrost, VoidBot, Hermodr) | `peer.flush` (`cultnet-ts/src/peer.ts:109-110`), then close | None. Cut 5 moves its connect-and-wait helper (`index.ts:6603-6650`) into cultnet-ts | | `cultmesh-ts/src/index.ts:6211-6303` `publishRudpDocumentOnce` (Heimdall, StreamPixels, Stonks, weksa, Bifrost, VoidBot, Hermodr) | `peer.flush` (`cultnet-ts/src/peer.ts:109-110`), then close | None. Cut 5 moves its connect-and-wait helper (`index.ts:6603-6650`) into cultnet-ts |
| `cultnet-ts/src/idunn-odin-presence-publisher.ts:89-215` `publishDocument`, behind `createIdunnRuntimePresencePublisher`. **StreamPixels' service runs this**: `apps/service/scripts/idunn-entry.mjs` runs `deployment/idunn/runtime-presence.mjs`, which calls `createIdunnRuntimePresencePublisher` (`:46`) from its vendored CultLib `542ddd3` (StreamPixels `a49f7d8`; pass 2's `30ee8b9` is stale), where it lives under its older name `cultnet-ts/src/idunn-runtime-authority.ts:206` | a hand-rolled session loop (`:186-208`) until `outstandingReliablePacketCount === 0`. It does resend (`:205`). It drops `readyToSend` (`:198`), so promoted queued packets wait for the resend timer. It pins `initialSequence: 1` (`:103`) and never sends a Disconnect, so Odin's hub keeps each publish's peer until it times out (each publish binds a new port, so each is a new peer) | Cut 5: the socket transport, `connectAndWait`, `send("schema")`, `flush`, `close` (which sends the Disconnect, `rudp.ts:956-969`) | | `cultnet-ts/src/idunn-odin-presence-publisher.ts:97-136` `sendPresenceDocument` (formerly `publishDocument`), behind `createIdunnRuntimePresencePublisher`. **StreamPixels' service runs this**: `apps/service/scripts/idunn-entry.mjs` runs `deployment/idunn/runtime-presence.mjs`, which calls `createIdunnRuntimePresencePublisher` (`:46`) from its vendored CultLib `542ddd3` (StreamPixels `a49f7d8`; pass 2's `30ee8b9` is stale), where it lives under its older name `cultnet-ts/src/idunn-runtime-authority.ts:206` | a hand-rolled session loop (`:186-208`) until `outstandingReliablePacketCount === 0`. It does resend (`:205`). It drops `readyToSend` (`:198`), so promoted queued packets wait for the resend timer. It pins `initialSequence: 1` (`:103`) and never sends a Disconnect, so Odin's hub keeps each publish's peer until it times out (each publish binds a new port, so each is a new peer) | Cut 5: the socket transport, `connectAndWait`, `send("schema")`, `flush`, `close` (which sends the Disconnect, `rudp.ts:956-969`) |
| `cultnet-ts/src/signed-daemon-health.ts:104-162` `publishSignedDaemonHealth` (Hermodr, Odin `idunn-rudp.cjs`) | a hand-rolled loop: sends the packets `sendMany` admitted, waits up to 500 ms for the **first** Ack packet of any content (`:146-153`), swallows the timeout, closes. Resends run inside that wait (`:303`) but promoted queued packets are never sent | Cut 5, as above. Today a write not acked within 500 ms, or any payload over 32 packets, is reported published | | `cultnet-ts/src/signed-daemon-health.ts:104-162` `publishSignedDaemonHealth` (Hermodr, Odin `idunn-rudp.cjs`) | a hand-rolled loop: sends the packets `sendMany` admitted, waits up to 500 ms for the **first** Ack packet of any content (`:146-153`), swallows the timeout, closes. Resends run inside that wait (`:303`) but promoted queued packets are never sent | Cut 5, as above. Today a write not acked within 500 ms, or any payload over 32 packets, is reported published |
| `cultnet-ts/src/operation-service.ts` | the operation response | none (the response is the proof) | | `cultnet-ts/src/operation-service.ts` | the operation response | none (the response is the proof) |
| C# `FlushReliable` callers: Gjallar `Program.cs:3055-3080` (two ordered frames, flush, dispose) | flush | None | | C# `FlushReliable` callers: Gjallar `Program.cs:3055-3080` (two ordered frames, flush, dispose) | flush | None |

File diff suppressed because it is too large Load diff

File diff suppressed because it is too large Load diff

View file

@ -5,8 +5,13 @@ All notable changes to `@gamecult/cultmesh-ts` are recorded here, following
## Unreleased ## Unreleased
### Added
- `publishRudpDocumentOnce` takes `writeReceipts: true`: it sends a hello listing `cultnet.write_admitted.v0` and `cultnet.error.v1` and, against a server that lists both back, resolves only on that write's `write_admitted`, rejects with `CultMeshRudpWriteRefusedError` (now also carrying an `error.v1`) on its refusal and with the new `CultMeshRudpWriteUnconfirmedError` (`reason`: `window_elapsed` after 5000 ms, `session_ended`, or `local_send_failed`) when nothing settles it. Without the option, or against a server that did not advertise, there is no hello and the window is 500 ms; a put nothing settled in it is unconfirmed too, no longer "published" (**behaviour change**: a server that applies a put but answers nothing now yields `CultMeshRudpWriteUnconfirmedError`, as the spec requires). A legacy server's echo of the put's id is not admission: only `write_admitted` is. The flush that sends the put is bounded by the result window, and a send not acknowledged by then is `local_send_failed`. `resultWindowMs` overrides either window. `publishRudpDocumentAndWaitForReceipt` is not changed.
### Breaking ### Breaking
- `publishRudpDocumentOnce` no longer treats a window with no refusal as published: a put nothing settled in its result window (500 ms, or 5000 ms against a receipt-capable server) rejects `CultMeshRudpWriteUnconfirmedError`, and a legacy server's echo of the put's id is not admission. It now resolves `"admitted"` (only on a `write_admitted`) or `"sent"`. **Migration:** a caller whose server cannot answer a write yet (for example Odin, until it is receipt-capable) passes `sendOnly: true` per call and treats `"sent"` as "sent", never as admitted. Receipts stay opt-in (`writeReceipts: true`); neither mode is the default.
- The package is renamed `cultmesh-ts` to `@gamecult/cultmesh-ts` (the repo already scopes `@gamecult/cultcache-ts`), so the unscoped name cannot be claimed by someone else. Import from `@gamecult/cultmesh-ts` and its subpaths; the API is unchanged. Depends on `@gamecult/cultnet-ts`. - The package is renamed `cultmesh-ts` to `@gamecult/cultmesh-ts` (the repo already scopes `@gamecult/cultcache-ts`), so the unscoped name cannot be claimed by someone else. Import from `@gamecult/cultmesh-ts` and its subpaths; the API is unchanged. Depends on `@gamecult/cultnet-ts`.
- `asSchemaAlias` is removed from `CultMeshDocumentHandle`, `CultMeshCollectionHandle` and their bound forms. Read the handle by the schema you mean. - `asSchemaAlias` is removed from `CultMeshDocumentHandle`, `CultMeshCollectionHandle` and their bound forms. Read the handle by the schema you mean.
- `CultMeshDocumentCatalog` (`document`, `tryDocument`, `latest`, `replace`, `submitPrediction`, `authoritativeWriter`, `predictionWriter`, `watch`) and `CultMeshCollectionCatalog` (`collection`, `tryCollection`, `latest`, `watchChanges`) take `<TDocument = never, TSchema>` and yield `CultMeshCatalogValue<TDocument, TSchema>`: an explicit type argument wins, a CultCache document definition names its own value type, and a bare schema descriptor yields `unknown`. A value type is no longer inferred from a value or an expected result. The `parse` option is removed from these calls, and a lookup by type answers with that type's handle or none, not an alias of a compatible schema. - `CultMeshDocumentCatalog` (`document`, `tryDocument`, `latest`, `replace`, `submitPrediction`, `authoritativeWriter`, `predictionWriter`, `watch`) and `CultMeshCollectionCatalog` (`collection`, `tryCollection`, `latest`, `watchChanges`) take `<TDocument = never, TSchema>` and yield `CultMeshCatalogValue<TDocument, TSchema>`: an explicit type argument wins, a CultCache document definition names its own value type, and a bare schema descriptor yields `unknown`. A value type is no longer inferred from a value or an expected result. The `parse` option is removed from these calls, and a lookup by type answers with that type's handle or none, not an alias of a compatible schema.

View file

@ -20,6 +20,10 @@ import {
CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportConnection,
rudpClientBindHost, rudpClientBindHost,
CultNetSchemaCatalog, CultNetSchemaCatalog,
CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS,
WriteTracker,
helloAdvertisesWriteReceipts,
resultWindowMs,
cultNetMessageInvalidError, cultNetMessageInvalidError,
cultNetTextError, cultNetTextError,
cultNetWriteNotAuthorizedError, cultNetWriteNotAuthorizedError,
@ -44,6 +48,8 @@ import {
type CultNetSchemaCatalogOptions, type CultNetSchemaCatalogOptions,
type CultNetSnapshotResponseRawMessage, type CultNetSnapshotResponseRawMessage,
type CultNetWireContract, type CultNetWireContract,
type WriteSettlement,
type WriteUnconfirmedReason,
} from "@gamecult/cultnet-ts"; } from "@gamecult/cultnet-ts";
import { import {
@ -4189,8 +4195,8 @@ export interface CultMeshRudpDocumentServer {
*/ */
export class CultMeshRudpWriteRefusedError extends Error { export class CultMeshRudpWriteRefusedError extends Error {
readonly code: string | undefined; readonly code: string | undefined;
readonly response: Extract<CultNetMessage, { schemaVersion: "cultnet.error.v0" }>; readonly response: Extract<CultNetMessage, { schemaVersion: "cultnet.error.v0" | "cultnet.error.v1" }>;
constructor(response: Extract<CultNetMessage, { schemaVersion: "cultnet.error.v0" }>) { constructor(response: Extract<CultNetMessage, { schemaVersion: "cultnet.error.v0" | "cultnet.error.v1" }>) {
super(response.error); super(response.error);
this.name = "CultMeshRudpWriteRefusedError"; this.name = "CultMeshRudpWriteRefusedError";
this.code = response.code ?? undefined; this.code = response.code ?? undefined;
@ -4198,14 +4204,49 @@ export class CultMeshRudpWriteRefusedError extends Error {
} }
} }
/**
* Nothing settled the write (write-receipt.md, "unconfirmed"): the result window ran out with no answer, the
* session ended first, or the client's own send failed. Not success and not refusal; a caller that needs admission treats it as not admitted.
*/
export class CultMeshRudpWriteUnconfirmedError extends Error {
readonly reason: WriteUnconfirmedReason;
readonly messageId: string;
constructor(messageId: string, reason: WriteUnconfirmedReason) {
super(`Write ${messageId} was not confirmed: ${reason}.`);
this.name = "CultMeshRudpWriteUnconfirmedError";
this.reason = reason;
this.messageId = messageId;
}
}
export interface CultMeshRudpDocumentPublishOptions extends CultMeshRudpPeerOptions { export interface CultMeshRudpDocumentPublishOptions extends CultMeshRudpPeerOptions {
/** /**
* How long after the put's transport ack a refusal may still arrive. A reliable-transport ack is not * How long, counted from the send, the put may wait for its result frame. A reliable-transport ack is not
* admission; the outcome is the peer's result frame. A server that answers a put only on refusal sends no * admission; the outcome is the peer's result frame. Default: 5000 ms against a server whose hello lists the
* success frame, so the put is reported published once this window passes with no refusal; a success reply * write-receipt frames, 500 ms otherwise. A put nothing settled within the window is unconfirmed, never
* ends the wait early. Default 500 ms. * published, whatever the server advertised: only a `write_admitted` is admission. The send itself is also
* bounded by the window (`flushTimeoutMs` is the upper bound); a send not acknowledged by then is
* unconfirmed as `local_send_failed`.
*/ */
resultWindowMs?: number; resultWindowMs?: number;
/**
* Declare the write-receipt capability: send a hello listing `cultnet.write_admitted.v0` and `cultnet.error.v1`
* before the put. A server that lists both back is receipt-capable: the put then resolves only on its
* `write_admitted`, rejects with `CultMeshRudpWriteRefusedError` on its `error.v1`, and rejects with
* `CultMeshRudpWriteUnconfirmedError` when nothing settles it. Unset, no hello is sent and the server answers
* as it always did.
*/
writeReceipts?: boolean;
/** With `writeReceipts`, how long to wait for the server's hello before treating it as legacy. Default 500 ms. */
helloWaitMs?: number;
/**
* Send the put and report only that it was sent: no hello, no wait for a result frame. Resolves `"sent"` once the
* transport has acknowledged the put (`flushTimeoutMs`), rejects `CultMeshRudpWriteUnconfirmedError`
* (`local_send_failed`) if it cannot be sent, and never resolves `"admitted"`: whether the server applied or
* refused it is unknown. For callers whose server cannot answer a write yet. Opt-in per call; not combinable with
* `writeReceipts`.
*/
sendOnly?: boolean;
messageId?: string; messageId?: string;
sourceRuntimeId?: string; sourceRuntimeId?: string;
sourceAgentId?: string; sourceAgentId?: string;
@ -6175,44 +6216,88 @@ export class CultMesh {
recordKey: string, recordKey: string,
value: CultCacheDocumentValue<TDefinition>, value: CultCacheDocumentValue<TDefinition>,
options: CultMeshRudpDocumentPublishOptions = {}, options: CultMeshRudpDocumentPublishOptions = {},
): Promise<void> { ): Promise<"admitted" | "sent"> {
requireNonEmpty(runtimeId, "runtimeId"); requireNonEmpty(runtimeId, "runtimeId");
requireNonEmpty(recordKey, "recordKey"); requireNonEmpty(recordKey, "recordKey");
if (options.sendOnly && options.writeReceipts) throw new TypeError("sendOnly and writeReceipts cannot be combined.");
const registry = new CultNetDocumentRegistry([binding]); const registry = new CultNetDocumentRegistry([binding]);
const messageId = options.messageId ?? `${runtimeId}:${binding.definition.type}:${recordKey}`; const messageId = options.messageId ?? `${runtimeId}:${binding.definition.type}:${recordKey}`;
// Built before any connection and outside every try: a put that cannot be built is the caller's error, with its cause.
const put = registry.createRawDocumentPutMessage(
binding,
messageId,
recordKey,
value,
{
sourceRuntimeId: options.sourceRuntimeId ?? runtimeId,
sourceAgentId: options.sourceAgentId,
sourceRole: options.sourceRole,
tags: options.tags,
},
);
const peer = await CultMesh.createRudpPeer(runtimeId, connectionId, endpoint, options); const peer = await CultMesh.createRudpPeer(runtimeId, connectionId, endpoint, options);
try { try {
// A transport ack is not admission: the put completes on its result frame. A refusal rejects with the if (options.sendOnly) {
// typed error whether it arrives before the ack, after it, or inside the result window. try {
let refusal: CultMeshRudpWriteRefusedError | undefined; peer.send(put);
let succeeded = false; await peer.flush(options.flushTimeoutMs ?? 30_000);
} catch {
throw new CultMeshRudpWriteUnconfirmedError(messageId, "local_send_failed");
}
return "sent";
}
// A transport ack is not admission: the put completes on its result frame, and the WriteTracker alone decides
// how it settles (write-receipt.md, "Client rules"). This function only maps the settlement to resolve/throw.
let receiptCapable = false;
let tracker: WriteTracker | undefined;
let settlement: WriteSettlement | undefined;
let wake: (() => void) | undefined; let wake: (() => void) | undefined;
let serverHello: (() => void) | undefined;
peer.on("message", (message: CultNetMessage) => { peer.on("message", (message: CultNetMessage) => {
if (message.schemaVersion === "cultnet.error.v0") refusal ??= new CultMeshRudpWriteRefusedError(message); if (message.schemaVersion === "cultnet.hello.v0") {
else if ("messageId" in message && message.messageId === messageId) succeeded = true; receiptCapable = helloAdvertisesWriteReceipts(message);
else return; serverHello?.();
wake?.(); } else {
tracker?.frame(message, Date.now());
}
}); });
peer.send(registry.createRawDocumentPutMessage( peer.on("close", () => tracker?.sessionEnd(Date.now()));
binding, if (options.writeReceipts) {
messageId, const answered = new Promise<void>((resolve) => { serverHello = resolve; });
recordKey, peer.sendHello({
value, schemaVersion: "cultnet.hello.v0", runtimeId, runtimeKind: "cultmesh-ts",
{ supportedMessageVersions: [...CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS],
sourceRuntimeId: options.sourceRuntimeId ?? runtimeId, });
sourceAgentId: options.sourceAgentId, let helloTimer: NodeJS.Timeout | undefined;
sourceRole: options.sourceRole, await Promise.race([answered, new Promise<void>((resolve) => { helloTimer = setTimeout(resolve, options.helloWaitMs ?? 500); })]);
tags: options.tags, clearTimeout(helloTimer);
}, }
)); const writeTracker = new WriteTracker({
await peer.flush(options.flushTimeoutMs ?? 30_000); windowMs: options.resultWindowMs ?? resultWindowMs(receiptCapable),
if (!refusal && !succeeded) { onSettle: (_id, result) => { settlement = result; wake?.(); },
});
tracker = writeTracker;
writeTracker.send(messageId, Date.now());
try {
peer.send(put);
// The flush may not outlast the result window: a send nobody acknowledged by then is a local send failure.
const windowRemaining = writeTracker.nextDeadlineMs(Date.now()) ?? 0;
await peer.flush(Math.min(options.flushTimeoutMs ?? 30_000, Math.max(windowRemaining, 1)));
} catch {
// Our own send failed or was never acknowledged: the put may or may not have left. Never a refusal.
writeTracker.sendFailed(messageId, Date.now());
}
while (!settlement) {
const remaining = writeTracker.nextDeadlineMs(Date.now()) ?? 0;
await new Promise<void>((resolve) => { await new Promise<void>((resolve) => {
const timer = setTimeout(resolve, options.resultWindowMs ?? 500); const timer = setTimeout(resolve, remaining);
wake = () => { clearTimeout(timer); resolve(); }; wake = () => { clearTimeout(timer); resolve(); };
}); });
writeTracker.tick(Date.now());
} }
if (refusal) throw refusal; if (settlement.outcome === "refused") throw new CultMeshRudpWriteRefusedError(settlement.response);
if (settlement.outcome === "unconfirmed") throw new CultMeshRudpWriteUnconfirmedError(messageId, settlement.reason);
return "admitted";
} finally { } finally {
peer.close(); peer.close();
} }

View file

@ -16,7 +16,7 @@ import {
encodeRudpPacket, encodeRudpPacket,
} from "@gamecult/cultnet-ts"; } from "@gamecult/cultnet-ts";
import { encodeWireValue } from "@gamecult/cultnet-ts/wire"; import { encodeWireValue } from "@gamecult/cultnet-ts/wire";
import { CultMesh, CultMeshGrantAllWrites, CultMeshRudpWriteRefusedError } from "../src/index"; import { CultMesh, CultMeshGrantAllWrites, CultMeshRudpWriteRefusedError, CultMeshRudpWriteUnconfirmedError } from "../src/index";
const isTheRefusal = (error: unknown): boolean => const isTheRefusal = (error: unknown): boolean =>
error instanceof CultMeshRudpWriteRefusedError && error.code === "write_not_authorized"; error instanceof CultMeshRudpWriteRefusedError && error.code === "write_not_authorized";
@ -100,7 +100,7 @@ for (const order of orders) {
}); });
} }
test("publishRudpDocumentOnce still publishes a put the server admitted", async () => { test("a put the server applied but did not answer is unconfirmed, never published", async () => {
const connectionId = nextConnection++; const connectionId = nextConnection++;
const puts: string[] = []; const puts: string[] = [];
const server = CultMesh.createRudpDocumentServer("completion-admit", connectionId, { const server = CultMesh.createRudpDocumentServer("completion-admit", connectionId, {
@ -111,7 +111,10 @@ test("publishRudpDocumentOnce still publishes a put the server admitted", async
}); });
await server.start(); await server.start();
try { try {
await CultMesh.publishRudpDocumentOnce("completion-admit-client", connectionId, `rudp://127.0.0.1:${server.bind.port}`, binding, "tests:completion:a", { noteId: "a", text: "x" }, { resendDelayMs: 50, resendPollMs: 5, resultWindowMs: 200 }); await assert.rejects(
assert.deepEqual(puts, ["tests:completion:a"]); CultMesh.publishRudpDocumentOnce("completion-admit-client", connectionId, `rudp://127.0.0.1:${server.bind.port}`, binding, "tests:completion:a", { noteId: "a", text: "x" }, { resendDelayMs: 50, resendPollMs: 5, resultWindowMs: 200 }),
(error: unknown) => error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "window_elapsed",
);
assert.deepEqual(puts, ["tests:completion:a"], "the server did apply it; this server just does not answer");
} finally { server.close(); } } finally { server.close(); }
}); });

View file

@ -21,7 +21,7 @@ import {
type CultNetMessage, type CultNetMessage,
type CultNetSnapshotResponseRawMessage, type CultNetSnapshotResponseRawMessage,
} from "@gamecult/cultnet-ts"; } from "@gamecult/cultnet-ts";
import { CultMesh, CultMeshGrantAllWrites, CultMeshRudpUnservableDocumentError } from "../src/index"; import { CultMesh, CultMeshGrantAllWrites, CultMeshRudpUnservableDocumentError, CultMeshRudpWriteUnconfirmedError } from "../src/index";
// A record's payload as the bytes a peer would put on the wire. // A record's payload as the bytes a peer would put on the wire.
const payloadOf = (value: unknown) => documentPayloadFromBytes(encodeWireValue(value)); const payloadOf = (value: unknown) => documentPayloadFromBytes(encodeWireValue(value));
@ -2968,25 +2968,30 @@ test("CultMesh TS publishes one registered document to a RUDP catalog", async ()
try { try {
await server.start(); await server.start();
await CultMesh.publishRudpDocumentOnce( // This server answers a put with nothing, so the writer cannot know the put landed: unconfirmed, though it did.
"cultmesh-ts-muninn", await assert.rejects(
connectionId, CultMesh.publishRudpDocumentOnce(
`rudp://127.0.0.1:${server.bind.port}`, "cultmesh-ts-muninn",
defineCultNetDocumentBinding({ definition: noteDocument }), connectionId,
"note:odin", `rudp://127.0.0.1:${server.bind.port}`,
{ defineCultNetDocumentBinding({ definition: noteDocument }),
noteId: "note:odin", "note:odin",
body: "pay respects once", {
}, noteId: "note:odin",
{ body: "pay respects once",
resendDelayMs: 25, },
resendPollMs: 5, {
maxFragmentBytes: 1024, resendDelayMs: 25,
maxPendingReliablePackets: 16, resendPollMs: 5,
connectTimeoutMs: 1_000, maxFragmentBytes: 1024,
flushTimeoutMs: 25, maxPendingReliablePackets: 16,
sourceRole: "test-provider", connectTimeoutMs: 1_000,
}, flushTimeoutMs: 25,
resultWindowMs: 100,
sourceRole: "test-provider",
},
),
(error: unknown) => error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "window_elapsed",
); );
await waitFor(() => received !== undefined, "published RUDP document"); await waitFor(() => received !== undefined, "published RUDP document");

View file

@ -0,0 +1,243 @@
// Client side of contracts/cultnet/interop/write-receipt.md in the one-shot writer: with `writeReceipts` the put
// completes only on a write_admitted carrying its id; a refusal, a silent window and a legacy server each end as the
// spec says. Each test serves the client a fake RUDP peer that acknowledges the put and answers as scripted.
import assert from "node:assert/strict";
import { createSocket, type Socket } from "node:dgram";
import test from "node:test";
import { z } from "zod";
import { defineDocumentType } from "@gamecult/cultcache-ts";
import {
CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS,
CultNetRudpSession,
decodeRudpPacket,
defineCultNetDocumentBinding,
encodeCultNetMessageForWire,
encodeRudpPacket,
type CultNetMessage,
} from "@gamecult/cultnet-ts";
import { decodeWireValue, encodeWireValue } from "@gamecult/cultnet-ts/wire";
import { CultMesh, CultMeshRudpWriteRefusedError, CultMeshRudpWriteUnconfirmedError } from "../src/index";
const note = defineDocumentType({
type: "cultmesh.receipt_note",
schemaId: "cultmesh.receipt_note.v0",
schema: z.object({ noteId: z.string(), text: z.string() }),
name: "noteId",
});
const binding = defineCultNetDocumentBinding({ definition: note });
let nextConnection = 0x10206000;
interface Script {
/** Does the server's hello list both receipt frames? */
advertise: boolean;
/** What the server answers a put with, `delayMs` after it arrives. */
answer: "admit" | "refuse" | "wrong-id" | "silence" | "goodbye" | "echo";
delayMs?: number;
/** Acknowledge the reliable packets it receives? Default true. */
ack?: boolean;
}
async function fakeServer(connectionId: number, script: Script) {
const socket: Socket = createSocket("udp4");
await new Promise<void>((resolve) => socket.bind(0, "127.0.0.1", resolve));
const session = new CultNetRudpSession({ connectionId, resendDelayMs: 50 });
const timers = new Set<NodeJS.Timeout>();
const seen: string[] = [];
let closed = false;
const send = (packet: Parameters<typeof encodeRudpPacket>[0], port: number, address: string) => {
if (!closed) socket.send(encodeRudpPacket(packet), port, address);
};
const say = (message: CultNetMessage, port: number, address: string) => {
const packets = session.sendMany("schema", encodeWireValue(encodeCultNetMessageForWire(message)), { reliable: true, ordered: true, nowMs: Date.now() });
packets.forEach((p) => send(p, port, address));
};
socket.on("message", (wire, remote) => {
const packet = decodeRudpPacket(wire);
const now = Date.now();
if (packet.packetType === "connect") {
send(session.acceptConnect(packet, now), remote.port, remote.address);
return;
}
const result = session.receive(packet, now);
if (result.reply) send(result.reply, remote.port, remote.address);
for (const frame of result.delivered) {
if (frame.channelId !== "schema") continue;
const message = decodeWireValue(frame.payload) as { schemaVersion: string; messageId?: string };
seen.push(message.schemaVersion);
if (message.schemaVersion === "cultnet.hello.v0") {
say({
schemaVersion: "cultnet.hello.v0", runtimeId: "fake", runtimeKind: "fake",
...(script.advertise ? { supportedMessageVersions: [...CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS] } : {}),
}, remote.port, remote.address);
continue;
}
if (message.schemaVersion !== "cultnet.document_put_raw.v0" || script.answer === "silence") continue;
const timer = setTimeout(() => {
timers.delete(timer);
if (script.answer === "echo") {
say({ schemaVersion: "cultnet.document_put_raw.v0", messageId: message.messageId!, document: { schemaId: "cultmesh.receipt_note.v0", recordKey: "tests:receipt:a", storedAt: "t", payloadEncoding: "messagepack", payload: new Uint8Array([1]) } } as unknown as CultNetMessage, remote.port, remote.address);
} else if (script.answer === "goodbye") {
send(session.endUnsendableSession(new Error("scripted goodbye")), remote.port, remote.address);
} else if (script.answer === "refuse") {
say({ schemaVersion: "cultnet.error.v1", error: "write_not_authorized", routingHint: null, code: "write_not_authorized", details: null, messageId: message.messageId! }, remote.port, remote.address);
} else {
const messageId = script.answer === "wrong-id" ? "someone-else" : message.messageId!;
say({ schemaVersion: "cultnet.write_admitted.v0", messageId, operation: "put", schemaId: "cultmesh.receipt_note.v0", recordKey: "tests:receipt:a" }, remote.port, remote.address);
}
}, script.delayMs ?? 20);
timers.add(timer);
}
if (packet.reliable && script.ack !== false) send(session.createAckForReceived(packet.sequence), remote.port, remote.address);
});
return { endpoint: `rudp://127.0.0.1:${socket.address().port}`, seen, close: () => { closed = true; timers.forEach(clearTimeout); socket.close(); } };
}
const transport = { resendDelayMs: 50, resendPollMs: 5, flushTimeoutMs: 2_000 };
const publish = (server: { endpoint: string }, options: Record<string, unknown>) =>
CultMesh.publishRudpDocumentOnce("receipt-client", nextConnection++, server.endpoint, binding, "tests:receipt:a", { noteId: "a", text: "x" }, { ...transport, ...options });
test("a receipt-capable server's write_admitted completes the put, past the legacy 500 ms window", async () => {
const server = await fakeServer(nextConnection, { advertise: true, answer: "admit", delayMs: 700 });
try {
await publish(server, { writeReceipts: true });
assert.deepEqual(server.seen.slice(0, 2), ["cultnet.hello.v0", "cultnet.document_put_raw.v0"]);
} finally { server.close(); }
});
test("a receipt-capable server's error.v1 refuses the put with its code", async () => {
const server = await fakeServer(nextConnection, { advertise: true, answer: "refuse" });
try {
await assert.rejects(publish(server, { writeReceipts: true }), (error: unknown) =>
error instanceof CultMeshRudpWriteRefusedError && error.code === "write_not_authorized" && error.response.schemaVersion === "cultnet.error.v1");
} finally { server.close(); }
});
test("a receipt-capable server that stays silent leaves the put unconfirmed, never published", async () => {
const server = await fakeServer(nextConnection, { advertise: true, answer: "silence" });
try {
await assert.rejects(publish(server, { writeReceipts: true, resultWindowMs: 250 }), (error: unknown) =>
error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "window_elapsed");
} finally { server.close(); }
});
test("a receipt for another id settles nothing", async () => {
const server = await fakeServer(nextConnection, { advertise: true, answer: "wrong-id" });
try {
await assert.rejects(publish(server, { writeReceipts: true, resultWindowMs: 250 }), CultMeshRudpWriteUnconfirmedError);
} finally { server.close(); }
});
test("a server whose hello does not advertise leaves a silent put unconfirmed after the 500 ms window, never published", async () => {
const server = await fakeServer(nextConnection, { advertise: false, answer: "silence" });
const started = Date.now();
try {
await assert.rejects(publish(server, { writeReceipts: true }), (error: unknown) =>
error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "window_elapsed");
assert.ok(Date.now() - started >= 500, "the legacy window ran from the send");
} finally { server.close(); }
});
test("a session that ends before any result leaves the put unconfirmed (session_ended), against a legacy server too", async () => {
for (const advertise of [false, true]) {
const server = await fakeServer(nextConnection, { advertise, answer: "goodbye" });
try {
await assert.rejects(publish(server, { writeReceipts: true, resultWindowMs: 3_000 }), (error: unknown) =>
error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "session_ended");
} finally { server.close(); }
}
});
test("a put whose send is never acknowledged is unconfirmed as local_send_failed, not an untyped error", async () => {
const server = await fakeServer(nextConnection, { advertise: true, answer: "silence", ack: false });
const started = Date.now();
try {
await assert.rejects(publish(server, { writeReceipts: true, flushTimeoutMs: 300 }), (error: unknown) =>
error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "local_send_failed");
assert.ok(Date.now() - started < 2_500, "flushTimeoutMs bounds the send inside the 5 s window");
} finally { server.close(); }
});
test("the flush is bounded by the result window: a send nobody acknowledges is local_send_failed at the window, not at the flush timeout", async () => {
const server = await fakeServer(nextConnection, { advertise: true, answer: "silence", ack: false });
const started = Date.now();
try {
await assert.rejects(publish(server, { writeReceipts: true, resultWindowMs: 300, flushTimeoutMs: 20_000 }), (error: unknown) =>
error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "local_send_failed");
assert.ok(Date.now() - started < 3_000, "it did not wait out the 20 s flush timeout");
} finally { server.close(); }
});
test("a server that does not advertise receipts and echoes the put's id back has not admitted it", async () => {
const server = await fakeServer(nextConnection, { advertise: false, answer: "echo" });
try {
await assert.rejects(publish(server, { writeReceipts: true, resultWindowMs: 300 }), (error: unknown) =>
error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "window_elapsed");
} finally { server.close(); }
});
test("a silent server never yields admitted, in either mode: the default is unconfirmed, sendOnly is the distinct outcome sent", async () => {
const dflt = await fakeServer(nextConnection, { advertise: false, answer: "silence" });
try {
await assert.rejects(publish(dflt, { resultWindowMs: 200 }), CultMeshRudpWriteUnconfirmedError);
} finally { dflt.close(); }
const sendOnly = await fakeServer(nextConnection, { advertise: false, answer: "silence" });
try {
const outcome = await publish(sendOnly, { sendOnly: true });
assert.equal(outcome, "sent");
assert.notEqual(outcome, "admitted");
assert.deepEqual(sendOnly.seen, ["cultnet.document_put_raw.v0"], "sendOnly sends no hello");
} finally { sendOnly.close(); }
});
test("only a write_admitted resolves admitted, and sendOnly does not wait for or act on an answer", async () => {
const admitting = await fakeServer(nextConnection, { advertise: true, answer: "admit" });
try {
assert.equal(await publish(admitting, { writeReceipts: true }), "admitted");
} finally { admitting.close(); }
const refusing = await fakeServer(nextConnection, { advertise: true, answer: "refuse", delayMs: 1 });
try {
assert.equal(await publish(refusing, { sendOnly: true }), "sent", "a refusal is not awaited: sendOnly says only that the put was sent");
} finally { refusing.close(); }
});
test("sendOnly fails local_send_failed when the send is never acknowledged", async () => {
const server = await fakeServer(nextConnection, { advertise: true, answer: "silence", ack: false });
try {
await assert.rejects(publish(server, { sendOnly: true, flushTimeoutMs: 300 }), (error: unknown) =>
error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "local_send_failed");
} finally { server.close(); }
});
test("sendOnly keeps the cause of a put that cannot be built: it is not reported as local_send_failed", async () => {
const server = await fakeServer(nextConnection, { advertise: false, answer: "silence" });
try {
await assert.rejects(
CultMesh.publishRudpDocumentOnce("receipt-client", nextConnection++, server.endpoint, binding, "tests:receipt:a", { noteId: 1 } as unknown as { noteId: string; text: string }, { ...transport, sendOnly: true }),
(error: unknown) => !(error instanceof CultMeshRudpWriteUnconfirmedError) && error instanceof Error,
);
assert.deepEqual(server.seen, [], "nothing was sent");
} finally { server.close(); }
});
test("the receipt path keeps the cause of a put that cannot be built: it is not reported as local_send_failed", async () => {
const server = await fakeServer(nextConnection, { advertise: true, answer: "admit" });
try {
await assert.rejects(
CultMesh.publishRudpDocumentOnce("receipt-client", nextConnection++, server.endpoint, binding, "tests:receipt:a", { noteId: 1 } as unknown as { noteId: string; text: string }, { ...transport, writeReceipts: true }),
(error: unknown) => !(error instanceof CultMeshRudpWriteUnconfirmedError) && error instanceof Error,
);
assert.ok(!server.seen.includes("cultnet.document_put_raw.v0"), "no put was sent");
} finally { server.close(); }
});
test("sendOnly cannot be combined with writeReceipts", async () => {
await assert.rejects(publish({ endpoint: "rudp://127.0.0.1:9" }, { sendOnly: true, writeReceipts: true }), TypeError);
});
test("without writeReceipts no hello is sent and the receipt frames are never asked for", async () => {
const server = await fakeServer(nextConnection, { advertise: true, answer: "silence" });
try {
await assert.rejects(publish(server, { resultWindowMs: 100 }), CultMeshRudpWriteUnconfirmedError);
assert.deepEqual(server.seen, ["cultnet.document_put_raw.v0"]);
} finally { server.close(); }
});

View file

@ -249,6 +249,9 @@ pub enum CultNetErrorCode {
WriteNotAuthorized, WriteNotAuthorized,
UnsupportedSchemaVersion, UnsupportedSchemaVersion,
MessageInvalid, MessageInvalid,
/// `cultnet.error.v1` only: the session's remembered-write limit is full. `cultnet.error.v0` neither
/// encodes nor decodes it, so its enum stays the schema's.
LimitReached,
} }
impl CultNetErrorCode { impl CultNetErrorCode {
@ -263,6 +266,7 @@ impl CultNetErrorCode {
Self::WriteNotAuthorized => "write_not_authorized", Self::WriteNotAuthorized => "write_not_authorized",
Self::UnsupportedSchemaVersion => "unsupported_schema_version", Self::UnsupportedSchemaVersion => "unsupported_schema_version",
Self::MessageInvalid => "message_invalid", Self::MessageInvalid => "message_invalid",
Self::LimitReached => "limit_reached",
} }
} }
@ -277,6 +281,7 @@ impl CultNetErrorCode {
"write_not_authorized" => Ok(Self::WriteNotAuthorized), "write_not_authorized" => Ok(Self::WriteNotAuthorized),
"unsupported_schema_version" => Ok(Self::UnsupportedSchemaVersion), "unsupported_schema_version" => Ok(Self::UnsupportedSchemaVersion),
"message_invalid" => Ok(Self::MessageInvalid), "message_invalid" => Ok(Self::MessageInvalid),
"limit_reached" => Ok(Self::LimitReached),
_ => Err(anyhow!("ErrorMessage.Code is not a recognized cultnet.error.v0 code")), _ => Err(anyhow!("ErrorMessage.Code is not a recognized cultnet.error.v0 code")),
} }
} }
@ -359,6 +364,26 @@ pub enum CultNetMessage {
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
details: Option<CultNetErrorDetails>, details: Option<CultNetErrorDetails>,
}, },
/// `cultnet.error.v1` (contracts/cultnet/interop/write-receipt.md): `cultnet.error.v0` plus the id of the
/// request it answers. Sent only to a connection that declared the write-receipt capability.
#[serde(skip_deserializing)]
#[serde(rename = "cultnet.error.v1", rename_all = "camelCase")]
ErrorV1 {
error: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
code: Option<CultNetErrorCode>,
#[serde(default, skip_serializing_if = "Option::is_none")]
details: Option<CultNetErrorDetails>,
message_id: Option<String>,
},
/// `cultnet.write_admitted.v0`: the receipt for one admitted put or delete, sent after its commit point.
#[serde(rename = "cultnet.write_admitted.v0", rename_all = "camelCase")]
WriteAdmitted {
message_id: String,
operation: String,
schema_id: String,
record_key: String,
},
#[serde(rename = "cultnet.sample.change_name.v0", rename_all = "camelCase")] #[serde(rename = "cultnet.sample.change_name.v0", rename_all = "camelCase")]
SampleChangeName { name: String }, SampleChangeName { name: String },
#[serde(rename = "cultnet.sample.chat.v0", rename_all = "camelCase")] #[serde(rename = "cultnet.sample.chat.v0", rename_all = "camelCase")]
@ -543,7 +568,9 @@ pub fn encode_cultnet_message_for_wire(
| CultNetMessage::SnapshotResponseRawV1 { .. } | CultNetMessage::SnapshotResponseRawV1 { .. }
| CultNetMessage::ContentChunkRequest { .. } | CultNetMessage::ContentChunkRequest { .. }
| CultNetMessage::ContentChunkResponse { .. } | CultNetMessage::ContentChunkResponse { .. }
| CultNetMessage::Error { .. } => encode_raw_cultnet_schema_message(message), | CultNetMessage::Error { .. }
| CultNetMessage::ErrorV1 { .. }
| CultNetMessage::WriteAdmitted { .. } => encode_raw_cultnet_schema_message(message),
_ => Ok(cultcache_rs::decode_payload(&rmp_serde::to_vec( _ => Ok(cultcache_rs::decode_payload(&rmp_serde::to_vec(
&serde_json::to_value(message)?, &serde_json::to_value(message)?,
)?)?), )?)?),
@ -573,7 +600,8 @@ pub fn decode_cultnet_message_from_slice(
| "cultnet.snapshot_response_raw.v0" | "cultnet.snapshot_response_raw.v0"
| "cultmesh.content_chunk_request.v1" | "cultmesh.content_chunk_request.v1"
| "cultmesh.content_chunk_response.v1" | "cultmesh.content_chunk_response.v1"
| "cultnet.error.v0", | "cultnet.error.v0"
| "cultnet.error.v1",
) => parse_raw_cultnet_schema_message(&wire_value) ) => parse_raw_cultnet_schema_message(&wire_value)
.with_context(|| received_version_label(&wire_value))?, .with_context(|| received_version_label(&wire_value))?,
_ => cultcache_rs::decode_payload::<CultNetMessage>(bytes) _ => cultcache_rs::decode_payload::<CultNetMessage>(bytes)
@ -645,6 +673,20 @@ fn validate_message(message: &CultNetMessage) -> Result<()> {
require_non_empty(session, "session")?; require_non_empty(session, "session")?;
} }
CultNetMessage::Error { error, code: _, details: _ } => require_non_empty(error, "error")?, CultNetMessage::Error { error, code: _, details: _ } => require_non_empty(error, "error")?,
CultNetMessage::ErrorV1 { error, message_id, .. } => {
require_non_empty(error, "error")?;
if let Some(message_id) = message_id {
require_non_empty(message_id, "messageId")?;
}
}
CultNetMessage::WriteAdmitted { message_id, operation, schema_id, record_key } => {
require_non_empty(message_id, "messageId")?;
if operation != "put" && operation != "delete" {
return Err(anyhow!("operation must be put or delete"));
}
require_non_empty(schema_id, "schemaId")?;
require_non_empty(record_key, "recordKey")?;
}
CultNetMessage::SampleChangeName { name } => require_non_empty(name, "name")?, CultNetMessage::SampleChangeName { name } => require_non_empty(name, "name")?,
CultNetMessage::SampleChat { text } => require_non_empty(text, "text")?, CultNetMessage::SampleChat { text } => require_non_empty(text, "text")?,
CultNetMessage::DocumentDelete { CultNetMessage::DocumentDelete {
@ -1021,7 +1063,7 @@ fn parse_raw_cultnet_schema_message(input: &rmpv::Value) -> Result<CultNetMessag
.to_vec(), .to_vec(),
error: require_legacy_string(get("error"), "error")?, error: require_legacy_string(get("error"), "error")?,
}), }),
"cultnet.error.v0" => { "cultnet.error.v0" | "cultnet.error.v1" => {
let code = get("code") let code = get("code")
.filter(|value| !value.is_nil()) .filter(|value| !value.is_nil())
.map(|value| { .map(|value| {
@ -1031,12 +1073,23 @@ fn parse_raw_cultnet_schema_message(input: &rmpv::Value) -> Result<CultNetMessag
.and_then(CultNetErrorCode::from_wire_str) .and_then(CultNetErrorCode::from_wire_str)
}) })
.transpose()?; .transpose()?;
if schema_version == "cultnet.error.v0" && code == Some(CultNetErrorCode::LimitReached) {
return Err(anyhow!("ErrorMessage.Code is not a recognized cultnet.error.v0 code"));
}
let details = require_error_details(get("details"), "ErrorMessage.Details")?; let details = require_error_details(get("details"), "ErrorMessage.Details")?;
Ok(CultNetMessage::Error { let error = require_legacy_string(get("error"), "ErrorMessage.Error")?;
error: require_legacy_string(get("error"), "ErrorMessage.Error")?, if schema_version == "cultnet.error.v1" {
code, return Ok(CultNetMessage::ErrorV1 {
details, error,
}) code,
details,
message_id: match get("messageId") {
None | Some(rmpv::Value::Nil) => None,
other => Some(require_legacy_string(other, "messageId")?),
},
});
}
Ok(CultNetMessage::Error { error, code, details })
} }
_ => Err(anyhow!( _ => Err(anyhow!(
"Unsupported raw cultnet.schema.v0 message" "Unsupported raw cultnet.schema.v0 message"
@ -1090,7 +1143,18 @@ fn require_legacy_optional_u64(
} }
fn encode_raw_cultnet_schema_message(message: &CultNetMessage) -> Result<rmpv::Value> { fn encode_raw_cultnet_schema_message(message: &CultNetMessage) -> Result<rmpv::Value> {
if let CultNetMessage::Error { code: Some(CultNetErrorCode::LimitReached), .. } = message {
return Err(anyhow!("limit_reached is a cultnet.error.v1 code; cultnet.error.v0 cannot carry it"));
}
Ok(rmpv::Value::Map(match message { Ok(rmpv::Value::Map(match message {
// Spec field order (write-receipt.md), not serde_json's sorted map.
CultNetMessage::WriteAdmitted { message_id, operation, schema_id, record_key } => vec![
(rmpv::Value::from("schemaVersion"), rmpv::Value::from("cultnet.write_admitted.v0")),
(rmpv::Value::from("messageId"), rmpv::Value::from(message_id.as_str())),
(rmpv::Value::from("operation"), rmpv::Value::from(operation.as_str())),
(rmpv::Value::from("schemaId"), rmpv::Value::from(schema_id.as_str())),
(rmpv::Value::from("recordKey"), rmpv::Value::from(record_key.as_str())),
],
CultNetMessage::DocumentPutRaw { CultNetMessage::DocumentPutRaw {
message_id, message_id,
document, document,
@ -1255,6 +1319,36 @@ fn encode_raw_cultnet_schema_message(message: &CultNetMessage) -> Result<rmpv::V
.unwrap_or(rmpv::Value::Nil), .unwrap_or(rmpv::Value::Nil),
), ),
], ],
CultNetMessage::ErrorV1 {
error,
code,
details,
message_id,
} => vec![
(rmpv::Value::from("schemaVersion"), rmpv::Value::from("cultnet.error.v1")),
(rmpv::Value::from("error"), rmpv::Value::from(error.as_str())),
(rmpv::Value::from("routingHint"), rmpv::Value::Nil),
(
rmpv::Value::from("code"),
code.as_ref()
.map(|code| rmpv::Value::from(code.as_wire_str()))
.unwrap_or(rmpv::Value::Nil),
),
(
rmpv::Value::from("details"),
details
.as_ref()
.map(encode_error_details)
.unwrap_or(rmpv::Value::Nil),
),
(
rmpv::Value::from("messageId"),
message_id
.as_deref()
.map(rmpv::Value::from)
.unwrap_or(rmpv::Value::Nil),
),
],
_ => { _ => {
return Err(anyhow!( return Err(anyhow!(
"Message is not a raw cultnet.schema.v0 binary replication message" "Message is not a raw cultnet.schema.v0 binary replication message"
@ -1884,3 +1978,71 @@ fn legacy_schema_descriptor(schema: &CultNetSchemaDescriptor) -> Result<rmpv::Va
), ),
])) ]))
} }
/// The content hash a write-receipt server binds to a `(actor, messageId)` dedupe key
/// (contracts/cultnet/interop/write-receipt.md, rule 5): SHA-256 of the MessagePack array
/// `[operation, schemaId, recordKey, payload]`, `payload` a `bin` for a put and nil for a delete.
pub fn write_content_hash(
operation: &str,
schema_id: &str,
record_key: &str,
payload: Option<&[u8]>,
) -> Result<[u8; 32]> {
use sha2::{Digest, Sha256};
let value = rmpv::Value::Array(vec![
rmpv::Value::from(operation),
rmpv::Value::from(schema_id),
rmpv::Value::from(record_key),
payload.map_or(rmpv::Value::Nil, |payload| rmpv::Value::Binary(payload.to_vec())),
]);
let mut bytes = Vec::new();
rmpv::encode::write_value(&mut bytes, &value)?;
Ok(Sha256::digest(&bytes).into())
}
/// SHA-256 of a write's `messageId`: what a receipt server remembers in place of the id itself, so an
/// entry's memory does not depend on the id's length (write-receipt.md, "What is remembered").
pub fn message_id_digest(message_id: &str) -> [u8; 32] {
use sha2::{Digest, Sha256};
Sha256::digest(message_id.as_bytes()).into()
}
/// The `messageId` of a schema frame that does not decode as a message (write-receipt.md, rule 3: a refusal
/// carries the write's id whenever the frame is readable enough to find it). `None` when the bytes are not
/// a MessagePack map or hold no non-empty string `messageId`.
pub fn readable_message_id(bytes: &[u8]) -> Option<String> {
let value = rmpv::decode::read_value(&mut &*bytes).ok()?;
let rmpv::Value::Map(entries) = value else {
return None;
};
entries.into_iter().find_map(|(key, value)| match (key.as_str(), value) {
(Some("messageId"), rmpv::Value::String(id)) => id.into_str().filter(|id| !id.is_empty()),
_ => None,
})
}
/// The first required field (`document`, then `schemaId`) that a put or delete frame lacks or holds as nil
/// (write-receipt.md, "Required fields"), with the write's operation (`"put"` or `"delete"`). `None` when the
/// bytes are not such a frame, or both fields are present. A put's `schemaId` is its document's.
pub fn missing_required_write_field(bytes: &[u8]) -> Option<(&'static str, &'static str)> {
let rmpv::Value::Map(entries) = rmpv::decode::read_value(&mut &*bytes).ok()? else {
return None;
};
let field = |entries: &[(rmpv::Value, rmpv::Value)], name: &str| {
entries
.iter()
.find(|(key, _)| key.as_str() == Some(name))
.map(|(_, value)| value.clone())
.filter(|value| !value.is_nil())
};
match field(&entries, "schemaVersion")?.as_str()? {
"cultnet.document_put_raw.v0" => match field(&entries, "document") {
Some(rmpv::Value::Map(document)) => {
field(&document, "schemaId").is_none().then_some(("put", "schemaId"))
}
_ => Some(("put", "document")),
},
"cultnet.document_delete.v0" => field(&entries, "schemaId").is_none().then_some(("delete", "schemaId")),
_ => None,
}
}

View file

@ -468,6 +468,14 @@ impl CultNetRudpSession {
self.create_disconnect(RUDP_REFUSED_PACKET_REASON.to_vec()) self.create_disconnect(RUDP_REFUSED_PACKET_REASON.to_vec())
} }
/// Ends a session because the server itself could not answer a write it owes an answer to (its answer
/// could not be encoded, queued or sent). The peer did nothing wrong: this goodbye is not the
/// "session refused a packet" of a protocol refusal.
pub fn end_local_failure_session(&mut self) -> CultNetRudpPacket {
self.reset_peer_state();
self.create_disconnect(RUDP_LOCAL_FAILURE_REASON.to_vec())
}
/// Ends a session that owes its peer a packet that can never be sent as /// Ends a session that owes its peer a packet that can never be sent as
/// built (see `is_permanent_send_error`). Resending it would fail forever, /// built (see `is_permanent_send_error`). Resending it would fail forever,
/// and dropping it would leave a reliable sequence the peer waits on for /// and dropping it would leave a reliable sequence the peer waits on for
@ -2609,6 +2617,8 @@ where
/// Sent to a peer whose session refused one of its packets, and reported to /// Sent to a peer whose session refused one of its packets, and reported to
/// the local caller as the disconnect reason. /// the local caller as the disconnect reason.
const RUDP_REFUSED_PACKET_REASON: &[u8] = b"session refused a packet"; const RUDP_REFUSED_PACKET_REASON: &[u8] = b"session refused a packet";
/// Goodbye reason of a session whose server could not answer a write (`end_local_failure_session`).
const RUDP_LOCAL_FAILURE_REASON: &[u8] = b"server could not answer a write";
const SESSION_TIMED_OUT_REASON: &[u8] = b"session timed out"; const SESSION_TIMED_OUT_REASON: &[u8] = b"session timed out";
/// Prefix of the goodbye reason for a session that owed a packet which can /// Prefix of the goodbye reason for a session that owed a packet which can
/// never be sent; the failure's fixed name (`send_error_code`) follows it. /// never be sent; the failure's fixed name (`send_error_code`) follows it.

View file

@ -0,0 +1,155 @@
//! Wire parity with contracts/cultnet/interop/write-receipt-vectors.json: the content hash, and the two new frames.
use cultnet_rs::{
CultNetErrorCode, CultNetErrorDetails, CultNetMessage, CultNetWireContract, decode_cultnet_message_from_slice,
encode_cultnet_message_to_vec, write_content_hash,
};
fn hex(bytes: &[u8]) -> String {
bytes.iter().map(|byte| format!("{byte:02x}")).collect()
}
#[test]
fn the_content_hash_matches_the_spec_vectors() {
let a = [0x81, 0xA2, 0x69, 0x64, 0xA1, 0x31];
let b = [0x81, 0xA2, 0x69, 0x64, 0xA1, 0x32];
let cases = [
(write_content_hash("put", "note.v0", "note:1", Some(&a)), "7051c10071123b6ade50fbd5075e3859cae4ea70ce358a24a0b0c6980090e95d"),
(write_content_hash("put", "note.v0", "note:1", Some(&b)), "58d3fc6b858d2605a68550c9d4811ccdca6b9db7a0c3bfebf304d279257f6f9f"),
(write_content_hash("put", "note.v0", "note:2", Some(&a)), "a248cd920e58d41bc6ccca9e108030b1d491239000047fef088122364a32b6f8"),
(write_content_hash("delete", "note.v0", "note:1", None), "232b0f96651b28aa8ab7673a6106944a9f6a8d66852b6e21291a23ed084b8b37"),
];
for (hash, expected) in cases {
assert_eq!(hex(&hash.unwrap()), expected);
}
}
#[test]
fn the_new_frames_decode_back_equal() {
let admitted = CultNetMessage::WriteAdmitted {
message_id: "put-1".into(),
operation: "put".into(),
schema_id: "note.v0".into(),
record_key: "note:1".into(),
};
let bytes = encode_cultnet_message_to_vec(&admitted, CultNetWireContract::CultNetSchemaV0).unwrap();
assert_eq!(
decode_cultnet_message_from_slice(&bytes, CultNetWireContract::CultNetSchemaV0).unwrap(),
admitted
);
// error.v1 with a nil id round-trips as nil, and the code enum is v0's.
let error = CultNetMessage::ErrorV1 {
error: "message_invalid: unreadable frame".into(),
code: Some(CultNetErrorCode::MessageInvalid),
details: None,
message_id: None,
};
let bytes = encode_cultnet_message_to_vec(&error, CultNetWireContract::CultNetSchemaV0).unwrap();
assert_eq!(
decode_cultnet_message_from_slice(&bytes, CultNetWireContract::CultNetSchemaV0).unwrap(),
error
);
}
fn unhex(text: &str) -> Vec<u8> {
(0..text.len()).step_by(2).map(|i| u8::from_str_radix(&text[i..i + 2], 16).unwrap()).collect()
}
/// The spec's wire vectors (write-receipt-vectors.json `wire`): the bytes decode to the vector's fields, not
/// merely to something that encodes back to itself.
#[test]
fn the_spec_wire_vectors_decode_to_their_expected_fields() {
let admitted = decode_cultnet_message_from_slice(
&unhex("85AD736368656D6156657273696F6EB963756C746E65742E77726974655F61646D69747465642E7630A96D6573736167654964A57075742D31A96F7065726174696F6EA3707574A8736368656D614964A76E6F74652E7630A97265636F72644B6579A66E6F74653A31"),
CultNetWireContract::CultNetSchemaV0,
)
.unwrap();
assert_eq!(
admitted,
CultNetMessage::WriteAdmitted {
message_id: "put-1".into(),
operation: "put".into(),
schema_id: "note.v0".into(),
record_key: "note:1".into(),
}
);
let refused = decode_cultnet_message_from_slice(
&unhex("86AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7631A56572726F72D93677726974655F6E6F745F617574686F72697A65643A2074686520636174616C6F6720726566757365642074686520646F63756D656E74AB726F7574696E6748696E74C0A4636F6465B477726974655F6E6F745F617574686F72697A6564A764657461696C7384A56669656C64A97265636F72644B6579A576616C7565A66E6F74653A32A461734F66C0A763757272656E74C0A96D6573736167654964A57075742D32"),
CultNetWireContract::CultNetSchemaV0,
)
.unwrap();
let CultNetMessage::ErrorV1 { error, code, details, message_id } = refused else {
panic!("expected error.v1");
};
assert_eq!(error, "write_not_authorized: the catalog refused the document");
assert_eq!(code, Some(CultNetErrorCode::WriteNotAuthorized));
assert_eq!(message_id.as_deref(), Some("put-2"));
let details = details.unwrap();
assert_eq!(details.field.as_deref(), Some("recordKey"));
assert_eq!(details.value.as_deref(), Some("note:2"));
assert_eq!((details.as_of, details.current), (None, None));
}
/// Spec vector `error_v1_limit_reached`: the bytes the server must send, and what they decode to.
#[test]
fn error_v1_limit_reached_is_the_spec_vector() {
let vector = "86AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7631A56572726F72D93B6C696D69745F726561636865643A207468652073657373696F6E27732072656D656D62657265642D7772697465206C696D69742069732066756C6CAB726F7574696E6748696E74C0A4636F6465AD6C696D69745F72656163686564A764657461696C7384A56669656C64C0A576616C7565C0A461734F66C0A763757272656E74C0A96D6573736167654964A87075742D31303235";
let message = CultNetMessage::ErrorV1 {
error: "limit_reached: the session's remembered-write limit is full".into(),
code: Some(CultNetErrorCode::LimitReached),
details: Some(CultNetErrorDetails::default()),
message_id: Some("put-1025".into()),
};
let bytes = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap();
assert_eq!(bytes, unhex(vector), "encoded bytes must equal the spec vector");
let CultNetMessage::ErrorV1 { error, code, details, message_id } =
decode_cultnet_message_from_slice(&unhex(vector), CultNetWireContract::CultNetSchemaV0).unwrap()
else {
panic!("expected error.v1");
};
assert_eq!(error, "limit_reached: the session's remembered-write limit is full");
assert_eq!(code, Some(CultNetErrorCode::LimitReached));
assert_eq!(message_id.as_deref(), Some("put-1025"));
assert_eq!(details, Some(CultNetErrorDetails::default()));
}
/// Spec vectors `write_admitted_put` and `write_admitted_delete`: the encoder emits the spec's field order.
#[test]
fn write_admitted_encodes_to_the_spec_vector_bytes() {
let cases = [
("put-1", "put", "85AD736368656D6156657273696F6EB963756C746E65742E77726974655F61646D69747465642E7630A96D6573736167654964A57075742D31A96F7065726174696F6EA3707574A8736368656D614964A76E6F74652E7630A97265636F72644B6579A66E6F74653A31"),
("del-1", "delete", "85AD736368656D6156657273696F6EB963756C746E65742E77726974655F61646D69747465642E7630A96D6573736167654964A564656C2D31A96F7065726174696F6EA664656C657465A8736368656D614964A76E6F74652E7630A97265636F72644B6579A66E6F74653A31"),
];
for (id, operation, vector) in cases {
let message = CultNetMessage::WriteAdmitted {
message_id: id.into(),
operation: operation.into(),
schema_id: "note.v0".into(),
record_key: "note:1".into(),
};
let bytes = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap();
assert_eq!(bytes, unhex(vector));
}
}
/// `limit_reached` is in the error.v1 enum only: a v0 error neither encodes nor decodes it.
#[test]
fn error_v0_refuses_the_v1_only_limit_reached_code() {
let v0 = CultNetMessage::Error {
error: "x".into(),
code: Some(CultNetErrorCode::LimitReached),
details: None,
};
assert!(encode_cultnet_message_to_vec(&v0, CultNetWireContract::CultNetSchemaV0).is_err());
let v1 = CultNetMessage::ErrorV1 {
error: "x".into(),
code: Some(CultNetErrorCode::LimitReached),
details: None,
message_id: None,
};
let v1_bytes = encode_cultnet_message_to_vec(&v1, CultNetWireContract::CultNetSchemaV0).unwrap();
let at = v1_bytes.windows(16).position(|w| w == b"cultnet.error.v1").unwrap();
let mut as_v0 = v1_bytes.clone();
as_v0[at + 15] = b'0';
assert!(decode_cultnet_message_from_slice(&as_v0, CultNetWireContract::CultNetSchemaV0).is_err());
}

View file

@ -7,6 +7,7 @@ All notable changes to `@gamecult/cultnet-ts` are recorded here, following
### Added ### Added
- Write receipt, client side (`contracts/cultnet/interop/write-receipt.md`): `cultnet.write_admitted.v0` and `cultnet.error.v1` (v0 plus `messageId`; its closed code set adds `limit_reached`, `CULTNET_ERROR_V1_CODES`) parse and encode, and `CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS` / `helloAdvertisesWriteReceipts` name the hello capability. `cultnet.error.v0` and its code set are unchanged. `WriteTracker` settles each write as admitted, refused or unconfirmed (windows `RECEIPT_RESULT_WINDOW_MS` 5000 and `LEGACY_RESULT_WINDOW_MS` 500, counted from the send; late answers recorded as `admitted_late` / `refused_late`), and `tools/replay-write-receipt-vectors.mjs` replays the vectors against it (see README, "Write receipt replay runner"). The interop schemas are read from `contracts/cultnet/interop/`.
- `message_invalid` joins the closed `cultnet.error.v0` code set, with `cultNetMessageInvalidError(field, reason)` (`details.field` names the request field at fault), byte-identical to the C# `ForMessageInvalid`. - `message_invalid` joins the closed `cultnet.error.v0` code set, with `cultNetMessageInvalidError(field, reason)` (`details.field` names the request field at fault), byte-identical to the C# `ForMessageInvalid`.
- `cultNetWriteNotAuthorizedError(recordKey, reason?)`: the typed `write_not_authorized` refusal (code and `details.recordKey`), byte-identical to the C# `ForRemoteWriteRefused` vector in `error-vectors.json`. `CultNetDocumentRegistry.canonicalSchemaId(schemaId)` resolves the id a put or delete writes under, for a server's write authority. - `cultNetWriteNotAuthorizedError(recordKey, reason?)`: the typed `write_not_authorized` refusal (code and `details.recordKey`), byte-identical to the C# `ForRemoteWriteRefused` vector in `error-vectors.json`. `CultNetDocumentRegistry.canonicalSchemaId(schemaId)` resolves the id a put or delete writes under, for a server's write authority.
- `cultNetTextError(error)`: the text-only `cultnet.error.v0` refusal, with `routingHint`, `code` and `details` null. The legacy-union error decode uses it, so there is one place that spells the null members. - `cultNetTextError(error)`: the text-only `cultnet.error.v0` refusal, with `routingHint`, `code` and `details` null. The legacy-union error decode uses it, so there is one place that spells the null members.
@ -16,6 +17,7 @@ All notable changes to `@gamecult/cultnet-ts` are recorded here, following
### Breaking ### Breaking
- The Idunn runtime presence publisher is send-only (`sendPresenceDocument`): it returns `"sent"` once Odin's transport has acknowledged the put and no longer waits up to 500 ms for a success reply, which was never admission. It rejects `IdunnPresenceRefusedError` for a `cultnet.error.v0`, or a `cultnet.error.v1` naming the put or nothing, that arrives by the ack; a refusal after the ack is not awaited, and no other frame, one carrying the put's id included, counts as success.
- The package is renamed `cultnet-ts` to `@gamecult/cultnet-ts` (the repo already scopes `@gamecult/cultcache-ts`), so the unscoped name cannot be claimed by someone else. Import from `@gamecult/cultnet-ts` and its subpaths; the API is unchanged. - The package is renamed `cultnet-ts` to `@gamecult/cultnet-ts` (the repo already scopes `@gamecult/cultcache-ts`), so the unscoped name cannot be claimed by someone else. Import from `@gamecult/cultnet-ts` and its subpaths; the API is unchanged.
- `CultNetErrorMessage` carries `routingHint`, `code` and `details` (null when absent), as the reference writes them and `cultnet.error.schema.json` requires; the interface previously dropped them. New `CultNetErrorCode`, `CultNetErrorDetails` and `CultNetShardRoutingHint` types. - `CultNetErrorMessage` carries `routingHint`, `code` and `details` (null when absent), as the reference writes them and `cultnet.error.schema.json` requires; the interface previously dropped them. New `CultNetErrorCode`, `CultNetErrorDetails` and `CultNetShardRoutingHint` types.

View file

@ -289,6 +289,33 @@ maps them back to the exact C# union payload shape at the wire boundary.
serve and dial commands also exercise schema-v0 over the shared RUDP socket serve and dial commands also exercise schema-v0 over the shared RUDP socket
transport and advertise both transport profiles. transport and advertise both transport profiles.
## Write receipt replay runner
`WriteTracker` is the client half of `contracts/cultnet/interop/write-receipt.md`: a pure state machine on an abstract
clock that owns one decision, how each sent write settles. `tools/replay-write-receipt-vectors.mjs` replays
`write-receipt-vectors.json` against it (build first, then `node tools/replay-write-receipt-vectors.mjs`): it prints
`PASS`/`FAIL` per case and exits nonzero on any failure. Rust and C# port the same runner by implementing this interface
and replaying the same file.
Tracker interface (times are integer milliseconds on the vector's clock; the constructor takes the window):
| Call | Meaning |
| --- | --- |
| `new WriteTracker({ windowMs, onSettle })` | `windowMs` = 5000 when the server's hello lists both `cultnet.write_admitted.v0` and `cultnet.error.v1`, else 500. `onSettle(id, settlement)` fires once per write. |
| `send(id, atMs)` | A write with that `messageId` was sent. Its window runs from here, not from any transport ack. |
| `frame(message, atMs)` | A decoded frame arrived. `write_admitted` and `error.v1` settle the write they name; `error.v0`, or `error.v1` with nil `messageId`, refuses a write only when exactly one is unanswered; an id not in flight settles nothing. |
| `tick(atMs)` | Advance the clock. Every unanswered write with `atMs - sentAt >= windowMs` settles `unconfirmed(window_elapsed)`. `send`, `frame` and `sessionEnd` tick first. |
| `sessionEnd(atMs)` | Every unanswered write settles `unconfirmed(session_ended)`. |
| `state(id)` | `pending`, or the settlement plus `recorded`: `admitted_late` / `refused_late`, set by the first answer that arrives after `unconfirmed`. A settlement never changes. |
Runner, per case in `scenarios.cases` (the transport ack is not an input: delivery is not an answer):
1. The fake server states its capability the only way a real one does, in a hello (`server: "receipt"` lists both frames, `"legacy"` lists none); the window comes from that hello and must equal `constants[case.window]`.
2. Events run in `atMs` order: `send` -> `send(id)`, `frame` -> encode the frame to MessagePack, decode and validate it as the client would, then `frame(message)`, `tick` -> `tick`, `sessionEnd` -> `sessionEnd`, `transportAck` -> nothing.
3. For each id in `expect`, the settlement given to the caller has the expected `outcome` (and `code` or `reason` when given), `state(id).recorded` equals the expected `recorded` (absent when absent), and nothing settled twice or outside `expect`.
The runner also decodes each `wire[].messagePackHex` and compares the fields and the re-encoded bytes to the vector, and requires every `schemaInvalid[]` frame to be refused by the decoder. `dedupe.cases` and `contentHash` are server vectors and not replayed here.
## Development ## Development
```bash ```bash

View file

@ -9,6 +9,8 @@ import registerSchema from "../contracts/cultnet/cultnet.register.schema.json";
import verifySchema from "../contracts/cultnet/cultnet.verify.schema.json"; import verifySchema from "../contracts/cultnet/cultnet.verify.schema.json";
import loginSuccessSchema from "../contracts/cultnet/cultnet.login-success.schema.json"; import loginSuccessSchema from "../contracts/cultnet/cultnet.login-success.schema.json";
import errorSchema from "../contracts/cultnet/cultnet.error.schema.json"; import errorSchema from "../contracts/cultnet/cultnet.error.schema.json";
import errorV1Schema from "../contracts/cultnet/interop/cultnet.error.v1.schema.json";
import writeAdmittedSchema from "../contracts/cultnet/interop/cultnet.write-admitted.schema.json";
import sampleChangeNameSchema from "../contracts/cultnet/cultnet.sample-change-name.schema.json"; import sampleChangeNameSchema from "../contracts/cultnet/cultnet.sample-change-name.schema.json";
import sampleChatSchema from "../contracts/cultnet/cultnet.sample-chat.schema.json"; import sampleChatSchema from "../contracts/cultnet/cultnet.sample-chat.schema.json";
import documentPutSchema from "../contracts/cultnet/cultnet.document-put.schema.json"; import documentPutSchema from "../contracts/cultnet/cultnet.document-put.schema.json";
@ -62,6 +64,8 @@ export type CultNetSchemaVersion =
| "cultnet.verify.v0" | "cultnet.verify.v0"
| "cultnet.login_success.v0" | "cultnet.login_success.v0"
| "cultnet.error.v0" | "cultnet.error.v0"
| "cultnet.error.v1"
| "cultnet.write_admitted.v0"
| "cultnet.sample.change_name.v0" | "cultnet.sample.change_name.v0"
| "cultnet.sample.chat.v0" | "cultnet.sample.chat.v0"
| "cultnet.document_put.v0" | "cultnet.document_put.v0"
@ -201,6 +205,11 @@ export const CULTNET_ERROR_CODES = [
export type CultNetErrorCode = (typeof CULTNET_ERROR_CODES)[number]; export type CultNetErrorCode = (typeof CULTNET_ERROR_CODES)[number];
/** The closed refusal-code set of cultnet.error.v1 (contracts/cultnet/interop/cultnet.error.v1.schema.json): v0's codes plus `limit_reached`. */
export const CULTNET_ERROR_V1_CODES = [...CULTNET_ERROR_CODES, "limit_reached"] as const;
export type CultNetErrorV1Code = (typeof CULTNET_ERROR_V1_CODES)[number];
/** A refusal that carries text only: `routingHint`, `code` and `details` are written as null, as the reference writes an absent member. */ /** A refusal that carries text only: `routingHint`, `code` and `details` are written as null, as the reference writes an absent member. */
export function cultNetTextError(error: string): CultNetErrorMessage { export function cultNetTextError(error: string): CultNetErrorMessage {
return { schemaVersion: "cultnet.error.v0", error, routingHint: null, code: null, details: null }; return { schemaVersion: "cultnet.error.v0", error, routingHint: null, code: null, details: null };
@ -255,6 +264,34 @@ export interface CultNetErrorMessage {
details: CultNetErrorDetails | null; details: CultNetErrorDetails | null;
} }
/**
* cultnet.error.v0 plus the `messageId` of the request it answers (nil when no id could be read). Sent only to a
* connection whose hello listed both receipt frames (contracts/cultnet/interop/write-receipt.md).
*/
export interface CultNetErrorV1Message extends Omit<CultNetErrorMessage, "schemaVersion" | "code"> {
schemaVersion: "cultnet.error.v1";
code: CultNetErrorV1Code | null;
messageId: string | null;
}
/** The receipt for one admitted put or delete: every key written, the write's `messageId` echoed byte for byte. */
export interface CultNetWriteAdmittedMessage {
schemaVersion: "cultnet.write_admitted.v0";
messageId: string;
operation: "put" | "delete";
schemaId: string;
recordKey: string;
}
/** The two frames a receipt-capable hello lists in `supportedMessageVersions`; a peer lists both or neither. */
export const CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS = ["cultnet.write_admitted.v0", "cultnet.error.v1"] as const;
/** True when a hello lists both receipt frames: the only thing that makes a connection receipt-capable. */
export function helloAdvertisesWriteReceipts(hello: Pick<CultNetHelloMessage, "supportedMessageVersions">): boolean {
const listed = hello.supportedMessageVersions ?? [];
return CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS.every((version) => listed.includes(version));
}
export interface CultNetSampleChangeNameMessage { export interface CultNetSampleChangeNameMessage {
schemaVersion: "cultnet.sample.change_name.v0"; schemaVersion: "cultnet.sample.change_name.v0";
name: string; name: string;
@ -770,6 +807,8 @@ export interface GhostlightAgentStateDocument {
export type CultNetMessage = export type CultNetMessage =
| CultNetHelloMessage | CultNetHelloMessage
| CultNetErrorV1Message
| CultNetWriteAdmittedMessage
| CultNetLoginMessage | CultNetLoginMessage
| CultNetRegisterMessage | CultNetRegisterMessage
| CultNetVerifyMessage | CultNetVerifyMessage
@ -812,6 +851,8 @@ const CULTNET_MESSAGE_SCHEMAS = [
verifySchema, verifySchema,
loginSuccessSchema, loginSuccessSchema,
errorSchema, errorSchema,
errorV1Schema,
writeAdmittedSchema,
sampleChangeNameSchema, sampleChangeNameSchema,
sampleChatSchema, sampleChatSchema,
documentPutSchema, documentPutSchema,
@ -1535,6 +1576,8 @@ export const cultNetSchemas = {
verifySchema, verifySchema,
loginSuccessSchema, loginSuccessSchema,
errorSchema, errorSchema,
errorV1Schema,
writeAdmittedSchema,
sampleChangeNameSchema, sampleChangeNameSchema,
sampleChatSchema, sampleChatSchema,
documentPutSchema, documentPutSchema,

View file

@ -52,7 +52,7 @@ export function createIdunnRuntimePresencePublisher(
const publish = async (detail: string) => { const publish = async (detail: string) => {
const signed = signer.sign(detail); const signed = signer.sign(detail);
messageSequence += 1; messageSequence += 1;
await publishDocument(endpoint, connectionId, { await sendPresenceDocument(endpoint, connectionId, {
schemaVersion: "cultnet.document_put_raw.v0", schemaVersion: "cultnet.document_put_raw.v0",
messageId: `runtime-presence:${signer.authority.expected.target}:${signer.runtimeInstanceId}:${messageSequence}`, messageId: `runtime-presence:${signer.authority.expected.target}:${signer.runtimeInstanceId}:${messageSequence}`,
document: signed.document, document: signed.document,
@ -86,11 +86,19 @@ export function createIdunnRuntimePresencePublisher(
}; };
} }
async function publishDocument( /**
* Sends one presence put and says only that it was sent (write-receipt.md: the explicit send-only mode). Odin does not
* answer a write with a receipt yet, so no frame, whatever id it carries, is admission; this never claims it. It returns
* once the transport has acknowledged the put, rejects `IdunnPresenceRefusedError` if a refusal arrives by then, and
* rejects if the put is never acknowledged.
*
* @internal exported for tests
*/
export async function sendPresenceDocument(
endpoint: { host: string; port: number }, endpoint: { host: string; port: number },
connectionId: number, connectionId: number,
message: Record<string, unknown>, message: Record<string, unknown>,
): Promise<void> { ): Promise<"sent"> {
const socket = dgram.createSocket(endpoint.host.includes(":") ? "udp6" : "udp4"); const socket = dgram.createSocket(endpoint.host.includes(":") ? "udp6" : "udp4");
await new Promise<void>((resolve, reject) => { await new Promise<void>((resolve, reject) => {
socket.once("error", reject); socket.once("error", reject);
@ -107,26 +115,24 @@ async function publishDocument(
const wirePayload = encode(encodeCultNetMessageForWire(message as never, "cultnet.schema.v0")); const wirePayload = encode(encodeCultNetMessageForWire(message as never, "cultnet.schema.v0"));
const packets = session.sendMany("schema", wirePayload, { reliable: true, ordered: true, nowMs: Date.now() }); const packets = session.sendMany("schema", wirePayload, { reliable: true, ordered: true, nowMs: Date.now() });
for (const packet of packets) await sendPacket(socket, endpoint, packet); for (const packet of packets) await sendPacket(socket, endpoint, packet);
// The transport ack is not admission: the put completes on Odin's result frame. A refusal rejects whether // The transport ack is delivery, not admission. A refusal that arrives by then rejects; no other frame means anything.
// it arrives before the ack, after it, or inside the result window; a success reply ends the wait early.
let succeeded = false;
const onFrame = (frame: { channelId: string; payload: Uint8Array }) => { const onFrame = (frame: { channelId: string; payload: Uint8Array }) => {
if (frame.channelId !== "schema") return; if (frame.channelId !== "schema") return;
const response = parseCultNetMessage(decode(frame.payload)); const response = parseCultNetMessage(decode(frame.payload));
if (response.schemaVersion === "cultnet.error.v0") throw new IdunnPresenceRefusedError(response); if (response.schemaVersion === "cultnet.error.v0") throw new IdunnPresenceRefusedError(response);
succeeded = true; // An error.v1 refuses this put only when it names it (or names nothing).
if (response.schemaVersion === "cultnet.error.v1" && (response.messageId === null || response.messageId === message.messageId)) {
throw new IdunnPresenceRefusedError(response);
}
}; };
await receiveUntil(receiver, session, endpoint, () => session.outstandingReliablePacketCount === 0, 2000, onFrame); await receiveUntil(receiver, session, endpoint, () => session.outstandingReliablePacketCount === 0, 2000, onFrame);
if (!succeeded) await receiveUntil(receiver, session, endpoint, () => succeeded, RESULT_WINDOW_MS, onFrame, true); return "sent";
} finally { } finally {
receiver.close(); receiver.close();
socket.close(); socket.close();
} }
} }
/** How long after the transport ack Odin's refusal may still arrive; a success reply ends the wait early. */
const RESULT_WINDOW_MS = 500;
/** Odin refused the presence put: the typed `cultnet.error.v0` it answered with. */ /** Odin refused the presence put: the typed `cultnet.error.v0` it answered with. */
export class IdunnPresenceRefusedError extends Error { export class IdunnPresenceRefusedError extends Error {
readonly code: string | undefined; readonly code: string | undefined;
@ -200,7 +206,6 @@ async function receiveUntil(
predicate: (packet: CultNetRudpPacket) => boolean, predicate: (packet: CultNetRudpPacket) => boolean,
timeoutMs: number, timeoutMs: number,
onDelivered?: (frame: { channelId: string; payload: Uint8Array; sequence: number }) => void, onDelivered?: (frame: { channelId: string; payload: Uint8Array; sequence: number }) => void,
returnOnTimeout = false,
): Promise<void> { ): Promise<void> {
const deadline = Date.now() + timeoutMs; const deadline = Date.now() + timeoutMs;
while (Date.now() < deadline) { while (Date.now() < deadline) {
@ -215,7 +220,6 @@ async function receiveUntil(
} }
for (const packet of session.dueResends(Date.now())) await sendPacket(receiver.socket, endpoint, packet); for (const packet of session.dueResends(Date.now())) await sendPacket(receiver.socket, endpoint, packet);
} }
if (returnOnTimeout) return;
throw new Error(`Timed out waiting for CultNet RUDP response after ${timeoutMs} ms.`); throw new Error(`Timed out waiting for CultNet RUDP response after ${timeoutMs} ms.`);
} }

View file

@ -17,6 +17,12 @@ export {
type CultNetRawDocumentRecord, type CultNetRawDocumentRecord,
type CultNetRawPayloadEncoding, type CultNetRawPayloadEncoding,
CULTNET_ERROR_CODES, CULTNET_ERROR_CODES,
CULTNET_ERROR_V1_CODES,
CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS,
helloAdvertisesWriteReceipts,
type CultNetErrorV1Code,
type CultNetErrorV1Message,
type CultNetWriteAdmittedMessage,
cultNetTextError, cultNetTextError,
cultNetMessageInvalidError, cultNetMessageInvalidError,
cultNetWriteNotAuthorizedError, cultNetWriteNotAuthorizedError,
@ -249,3 +255,15 @@ export {
type IdunnRuntimePresencePublisher, type IdunnRuntimePresencePublisher,
type IdunnRuntimePresencePublisherOptions, type IdunnRuntimePresencePublisherOptions,
} from "./idunn-odin-presence-publisher"; } from "./idunn-odin-presence-publisher";
export {
LEGACY_RESULT_WINDOW_MS,
RECEIPT_RESULT_WINDOW_MS,
WriteTracker,
resultWindowMs,
type WriteRecorded,
type WriteSettlement,
type WriteState,
type WriteTrackerOptions,
type WriteUnconfirmedReason,
} from "./write-tracker";

View file

@ -0,0 +1,144 @@
// The client half of the write receipt (contracts/cultnet/interop/write-receipt.md, "Client rules"): a pure state
// machine with an abstract clock. It owns exactly one decision, how each sent write settles. Transport, timers and
// frame decoding belong to the caller; the replay runner (tools/replay-write-receipt-vectors.mjs) drives this same
// interface from write-receipt-vectors.json.
import type { CultNetErrorMessage, CultNetErrorV1Message, CultNetMessage } from "./contracts";
/** The result window against a server whose hello lists both receipt frames. */
export const RECEIPT_RESULT_WINDOW_MS = 5000;
/** The result window against a server whose hello does not list them (or that sent none). */
export const LEGACY_RESULT_WINDOW_MS = 500;
export function resultWindowMs(serverAdvertisesReceipts: boolean): number {
return serverAdvertisesReceipts ? RECEIPT_RESULT_WINDOW_MS : LEGACY_RESULT_WINDOW_MS;
}
/**
* Why nothing settled a write: its window ran out, the session ended first, or the client's own send of it failed (it may or
* may not have left the machine, and is never a refusal).
*/
export type WriteUnconfirmedReason = "window_elapsed" | "session_ended" | "local_send_failed";
/** What a late answer is recorded as on a write that already settled unconfirmed. */
export type WriteRecorded = "admitted_late" | "refused_late";
/** What the caller was told. Settled once; never changed by a later frame. */
export type WriteSettlement =
| { outcome: "admitted" }
| { outcome: "refused"; code: string | null; error: string; response: CultNetErrorMessage | CultNetErrorV1Message }
| { outcome: "unconfirmed"; reason: WriteUnconfirmedReason };
/** A write's state: unanswered, or its settlement plus (after unconfirmed only) the late answer. */
export type WriteState =
| { outcome: "pending"; recorded?: undefined }
| (WriteSettlement & { recorded?: WriteRecorded });
export interface WriteTrackerOptions {
/** The result window, from the send. Use `resultWindowMs(serverAdvertisesReceipts)`. */
windowMs: number;
/** Called once per write, at the moment it settles. */
onSettle?: (messageId: string, settlement: WriteSettlement) => void;
}
interface Entry {
sentAtMs: number;
state: WriteState;
}
export class WriteTracker {
readonly #windowMs: number;
readonly #onSettle?: (messageId: string, settlement: WriteSettlement) => void;
readonly #writes = new Map<string, Entry>();
constructor(options: WriteTrackerOptions) {
this.#windowMs = options.windowMs;
this.#onSettle = options.onSettle;
}
/** Registers a write sent at `atMs`. A messageId is minted once per session: re-sending a known id is the same write. */
send(messageId: string, atMs: number): void {
this.tick(atMs);
if (!this.#writes.has(messageId)) this.#writes.set(messageId, { sentAtMs: atMs, state: { outcome: "pending" } });
}
/** Feeds one decoded frame received at `atMs`. Frames that are not answers to a write are ignored. */
frame(message: CultNetMessage, atMs: number): void {
this.tick(atMs);
switch (message.schemaVersion) {
case "cultnet.write_admitted.v0":
this.#answer(message.messageId, { outcome: "admitted" }, "admitted_late");
return;
case "cultnet.error.v1":
if (message.messageId !== null) {
this.#answer(message.messageId, refusalOf(message), "refused_late");
return;
}
break;
case "cultnet.error.v0":
break;
default:
return;
}
// An error that names no write is attributable only when exactly one write is unanswered.
const unanswered = [...this.#writes].filter(([, entry]) => entry.state.outcome === "pending");
if (unanswered.length === 1) this.#settle(unanswered[0][0], unanswered[0][1], refusalOf(message));
}
/** Advances the clock: every unanswered write whose window, counted from its send, has run settles unconfirmed. */
tick(atMs: number): void {
for (const [messageId, entry] of this.#writes) {
if (entry.state.outcome === "pending" && atMs - entry.sentAtMs >= this.#windowMs) {
this.#settle(messageId, entry, { outcome: "unconfirmed", reason: "window_elapsed" });
}
}
}
/** The session ended: every unanswered write settles unconfirmed. */
sessionEnd(atMs: number): void {
this.tick(atMs);
for (const [messageId, entry] of this.#writes) {
if (entry.state.outcome === "pending") this.#settle(messageId, entry, { outcome: "unconfirmed", reason: "session_ended" });
}
}
/** This client's own send of `messageId` failed: that write settles unconfirmed; the others are untouched. */
sendFailed(messageId: string, atMs: number): void {
const entry = this.#writes.get(messageId);
if (entry?.state.outcome === "pending") this.#settle(messageId, entry, { outcome: "unconfirmed", reason: "local_send_failed" });
this.tick(atMs);
}
state(messageId: string): WriteState | undefined {
return this.#writes.get(messageId)?.state;
}
/** Milliseconds from `atMs` until the next unanswered write's window runs out; undefined when none is unanswered. */
nextDeadlineMs(atMs: number): number | undefined {
let next: number | undefined;
for (const entry of this.#writes.values()) {
if (entry.state.outcome !== "pending") continue;
const remaining = Math.max(0, entry.sentAtMs + this.#windowMs - atMs);
if (next === undefined || remaining < next) next = remaining;
}
return next;
}
#answer(messageId: string, settlement: WriteSettlement, late: WriteRecorded): void {
const entry = this.#writes.get(messageId);
if (!entry) return; // names a write that is not in flight: settles nothing
if (entry.state.outcome === "pending") {
this.#settle(messageId, entry, settlement);
} else if (entry.state.outcome === "unconfirmed" && entry.state.recorded === undefined) {
entry.state = { ...entry.state, recorded: late };
}
}
#settle(messageId: string, entry: Entry, settlement: WriteSettlement): void {
entry.state = settlement;
this.#onSettle?.(messageId, settlement);
}
}
function refusalOf(message: CultNetErrorMessage | CultNetErrorV1Message): WriteSettlement {
return { outcome: "refused", code: message.code, error: message.error, response: message };
}

View file

@ -0,0 +1,94 @@
// The presence publisher is send-only (write-receipt.md): Odin answers no write with a receipt yet, so no frame, whatever
// id it carries, is admission. Each test serves it a fake Odin that acknowledges the put and answers as scripted.
import assert from "node:assert/strict";
import { createSocket, type Socket } from "node:dgram";
import test from "node:test";
import { encode, decode } from "@msgpack/msgpack";
import { encodeCultNetMessageForWire, type CultNetMessage } from "../src/contracts";
import { IdunnPresenceRefusedError, sendPresenceDocument } from "../src/idunn-odin-presence-publisher";
import { CultNetRudpSession, decodeRudpPacket, encodeRudpPacket } from "../src/rudp";
interface Script {
answer: "silence" | "echo" | "admitted" | "refuse-v0" | "refuse-v1-other" | "refuse-v1-same" | "refuse-v1-none";
/** Acknowledge the put? Default true. */
ack?: boolean;
}
async function fakeOdin(connectionId: number, script: Script) {
const socket: Socket = createSocket("udp4");
await new Promise<void>((resolve) => socket.bind(0, "127.0.0.1", resolve));
const session = new CultNetRudpSession({ connectionId, resendDelayMs: 50 });
const send = (packet: Parameters<typeof encodeRudpPacket>[0], port: number, address: string) => {
try { socket.send(encodeRudpPacket(packet), port, address); } catch { /* closed */ }
};
const say = (message: CultNetMessage, port: number, address: string) => {
const packets = session.sendMany("schema", encode(encodeCultNetMessageForWire(message, "cultnet.schema.v0")), { reliable: true, ordered: true, nowMs: Date.now() });
packets.forEach((p) => send(p, port, address));
};
socket.on("message", (wire, remote) => {
const packet = decodeRudpPacket(wire);
const now = Date.now();
if (packet.packetType === "connect") {
send(session.acceptConnect(packet, now), remote.port, remote.address);
return;
}
const result = session.receive(packet, now);
if (result.reply) send(result.reply, remote.port, remote.address);
for (const frame of result.delivered) {
if (frame.channelId !== "schema") continue;
const put = decode(frame.payload) as { messageId: string; document: unknown };
if (script.answer === "echo") say({ schemaVersion: "cultnet.document_put_raw.v0", messageId: put.messageId, document: put.document } as unknown as CultNetMessage, remote.port, remote.address);
else if (script.answer === "admitted") say({ schemaVersion: "cultnet.write_admitted.v0", messageId: put.messageId, operation: "put", schemaId: "s", recordKey: "k" }, remote.port, remote.address);
else if (script.answer === "refuse-v0") say({ schemaVersion: "cultnet.error.v0", error: "no", routingHint: null, code: "write_not_authorized", details: null }, remote.port, remote.address);
else if (script.answer === "refuse-v1-other") say({ schemaVersion: "cultnet.error.v1", error: "no", routingHint: null, code: "write_not_authorized", details: null, messageId: "someone-else" }, remote.port, remote.address);
else if (script.answer === "refuse-v1-same") say({ schemaVersion: "cultnet.error.v1", error: "no", routingHint: null, code: "write_not_authorized", details: null, messageId: put.messageId }, remote.port, remote.address);
else if (script.answer === "refuse-v1-none") say({ schemaVersion: "cultnet.error.v1", error: "no", routingHint: null, code: "write_not_authorized", details: null, messageId: null }, remote.port, remote.address);
}
if (packet.reliable && script.ack !== false) send(session.createAckForReceived(packet.sequence), remote.port, remote.address);
});
return { port: socket.address().port, close: () => socket.close() };
}
let nextConnection = 0x0d1d1000;
const put = (id: string) => ({ schemaVersion: "cultnet.document_put_raw.v0", messageId: id, document: { schemaId: "s", recordKey: "k", storedAt: "t", payloadEncoding: "messagepack", payload: new Uint8Array([1]) } });
const sendTo = (server: { port: number }, id: string, connectionId: number) =>
sendPresenceDocument({ host: "127.0.0.1", port: server.port }, connectionId, put(id));
for (const answer of ["silence", "echo", "admitted", "refuse-v1-other"] as const) {
test(`a ${answer} Odin yields sent, never admitted: no frame carrying the put's id is success`, async () => {
const connectionId = nextConnection++;
const server = await fakeOdin(connectionId, { answer });
try {
const outcome = await sendTo(server, "presence-1", connectionId);
assert.equal(outcome, "sent");
assert.notEqual(outcome as string, "admitted");
} finally { server.close(); }
});
}
test("a refusal that arrives by the transport ack rejects the put", async () => {
const connectionId = nextConnection++;
const server = await fakeOdin(connectionId, { answer: "refuse-v0" });
try {
await assert.rejects(sendTo(server, "presence-1", connectionId), IdunnPresenceRefusedError);
} finally { server.close(); }
});
for (const answer of ["refuse-v1-same", "refuse-v1-none"] as const) {
test(`an error.v1 refusal that ${answer === "refuse-v1-same" ? "names this put" : "names no put"} rejects the put`, async () => {
const connectionId = nextConnection++;
const server = await fakeOdin(connectionId, { answer });
try {
await assert.rejects(sendTo(server, "presence-1", connectionId), IdunnPresenceRefusedError);
} finally { server.close(); }
});
}
test("a put the transport never acknowledges is an error, not sent", async () => {
const connectionId = nextConnection++;
const server = await fakeOdin(connectionId, { answer: "silence", ack: false });
try {
await assert.rejects(sendTo(server, "presence-1", connectionId), /Timed out/);
} finally { server.close(); }
});

View file

@ -345,18 +345,27 @@ test("the Odin publisher publishes the signer's document to the endpoint Expecte
assert.equal(peer.failure, undefined, peer.failure?.message); assert.equal(peer.failure, undefined, peer.failure?.message);
}); });
for (const order of ["ack-then-refusal", "refusal-then-ack"] as const) { test("the Odin publisher rejects a presence whose refusal arrives before the transport ack", async (context) => {
test(`the Odin publisher does not complete on the ack of a refused presence (${order})`, async (context) => { const endpoint = expectedFacts("service").odinEndpoint!;
const endpoint = expectedFacts("service").odinEndpoint!; const peer = await startOdinPeer(context, endpoint);
const peer = await startOdinPeer(context, endpoint); peer.reject = true;
peer.reject = true; peer.order = "refusal-then-ack";
peer.order = order; const signer = signerFor(context, "service");
const signer = signerFor(context, "service"); signer.reportHealth("active");
signer.reportHealth("active"); const publisher = createIdunnRuntimePresencePublisher({ signer, endpoint });
const publisher = createIdunnRuntimePresencePublisher({ signer, endpoint }); await assert.rejects(publisher.publish("denied presence"), /Odin rejected runtime presence: test admission denied/);
await assert.rejects(publisher.publish("denied presence"), /Odin rejected runtime presence: test admission denied/); });
});
} test("the Odin publisher is send-only: a refusal that follows the transport ack is not awaited", async (context) => {
const endpoint = expectedFacts("service").odinEndpoint!;
const peer = await startOdinPeer(context, endpoint);
peer.reject = true;
peer.order = "ack-then-refusal";
const signer = signerFor(context, "service");
signer.reportHealth("active");
const publisher = createIdunnRuntimePresencePublisher({ signer, endpoint });
await publisher.publish("denied presence");
});
test("the Odin publisher leaves a capability shortfall to the Rust correlation instead of refusing to publish", async (context) => { test("the Odin publisher leaves a capability shortfall to the Rust correlation instead of refusing to publish", async (context) => {
const endpoint = expectedFacts("service").odinEndpoint!; const endpoint = expectedFacts("service").odinEndpoint!;

View file

@ -0,0 +1,229 @@
import test from "node:test";
import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { join } from "node:path";
import {
CULTNET_ERROR_CODES,
CULTNET_ERROR_V1_CODES,
CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS,
LEGACY_RESULT_WINDOW_MS,
RECEIPT_RESULT_WINDOW_MS,
WriteTracker,
cultNetSchemas,
helloAdvertisesWriteReceipts,
parseCultNetMessage,
resultWindowMs,
type CultNetErrorV1Message,
type CultNetMessage,
type CultNetWriteAdmittedMessage,
type WriteSettlement,
} from "../src";
const admitted = (messageId: string): CultNetWriteAdmittedMessage =>
({ schemaVersion: "cultnet.write_admitted.v0", messageId, operation: "put", schemaId: "note.v0", recordKey: "note:1" });
const refusedV1 = (messageId: string | null, code: CultNetErrorV1Message["code"] = "write_not_authorized"): CultNetErrorV1Message =>
({ schemaVersion: "cultnet.error.v1", error: "no", routingHint: null, code, details: null, messageId });
const errorV0 = (): CultNetMessage =>
({ schemaVersion: "cultnet.error.v0", error: "no", routingHint: null, code: "write_not_authorized", details: null });
test("error.v1 accepts every code of its closed enum, limit_reached included, and refuses any other", () => {
const schemaCodes = ((cultNetSchemas.errorV1Schema.properties.code as { enum: (string | null)[] }).enum).filter((c) => c !== null);
assert.deepEqual([...CULTNET_ERROR_V1_CODES].sort(), [...schemaCodes].sort());
for (const code of CULTNET_ERROR_V1_CODES) parseCultNetMessage({ ...refusedV1("w1"), code });
assert.throws(() => parseCultNetMessage({ ...refusedV1("w1"), code: "invented" }));
});
test("error.v1 echoes messageId byte for byte, allows nil, and requires the key", () => {
assert.equal((parseCultNetMessage(refusedV1("w-é 1")) as CultNetErrorV1Message).messageId, "w-é 1");
assert.equal((parseCultNetMessage(refusedV1(null)) as CultNetErrorV1Message).messageId, null);
const { messageId: _omitted, ...withoutId } = refusedV1("w1");
assert.throws(() => parseCultNetMessage(withoutId));
assert.throws(() => parseCultNetMessage(refusedV1("")));
});
test("error.v0 is unaffected: still closed, no messageId, no limit_reached", () => {
assert.deepEqual([...CULTNET_ERROR_CODES].includes("limit_reached" as never), false);
assert.throws(() => parseCultNetMessage({ ...errorV0(), code: "limit_reached" }));
assert.throws(() => parseCultNetMessage({ ...errorV0(), messageId: "w1" }));
assert.equal(parseCultNetMessage(errorV0()).schemaVersion, "cultnet.error.v0");
});
test("write_admitted needs all five keys, a known operation and no error key", () => {
parseCultNetMessage(admitted("w1"));
parseCultNetMessage({ ...admitted("w1"), operation: "delete" });
assert.throws(() => parseCultNetMessage({ ...admitted("w1"), operation: "upsert" }));
assert.throws(() => parseCultNetMessage({ ...admitted("w1"), error: "nope" }));
const { recordKey: _omitted, ...short } = admitted("w1");
assert.throws(() => parseCultNetMessage(short));
});
test("a hello is receipt-capable only when it lists both frames", () => {
assert.equal(helloAdvertisesWriteReceipts({ supportedMessageVersions: [...CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS] }), true);
assert.equal(helloAdvertisesWriteReceipts({ supportedMessageVersions: ["cultnet.write_admitted.v0"] }), false);
assert.equal(helloAdvertisesWriteReceipts({ supportedMessageVersions: ["cultnet.error.v1"] }), false);
assert.equal(helloAdvertisesWriteReceipts({}), false);
const hello = parseCultNetMessage({ schemaVersion: "cultnet.hello.v0", runtimeId: "r", runtimeKind: "k", supportedMessageVersions: [...CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS] });
assert.equal(hello.schemaVersion, "cultnet.hello.v0");
});
test("the window is 5000 ms for a receipt-capable server and 500 ms otherwise", () => {
assert.equal(resultWindowMs(true), RECEIPT_RESULT_WINDOW_MS);
assert.equal(resultWindowMs(false), LEGACY_RESULT_WINDOW_MS);
assert.equal(RECEIPT_RESULT_WINDOW_MS, 5000);
assert.equal(LEGACY_RESULT_WINDOW_MS, 500);
});
function tracker(windowMs = 500) {
const told: [string, WriteSettlement][] = [];
return { told, t: new WriteTracker({ windowMs, onSettle: (id, s) => told.push([id, s]) }) };
}
test("tracker: the window is inclusive at exactly windowMs and open one ms before", () => {
const { t, told } = tracker();
t.send("w1", 100);
t.tick(599);
assert.equal(t.state("w1")?.outcome, "pending");
t.tick(600);
assert.deepEqual(told, [["w1", { outcome: "unconfirmed", reason: "window_elapsed" }]]);
});
test("tracker: each write's window runs from its own send", () => {
const { t } = tracker();
t.send("w1", 0);
t.send("w2", 300);
t.tick(500);
assert.equal(t.state("w1")?.outcome, "unconfirmed");
assert.equal(t.state("w2")?.outcome, "pending");
});
test("tracker: a frame arriving after the window cannot admit the write; it is recorded late", () => {
const { t, told } = tracker();
t.send("w1", 0);
t.frame(admitted("w1"), 500);
assert.deepEqual(t.state("w1"), { outcome: "unconfirmed", reason: "window_elapsed", recorded: "admitted_late" });
assert.equal(told.length, 1);
});
test("tracker: a write settles once; later frames neither change it nor flip a late record", () => {
const { t, told } = tracker();
t.send("w1", 0);
t.frame(refusedV1("w1"), 10);
t.frame(admitted("w1"), 20);
t.frame(refusedV1("w1", "message_invalid"), 30);
t.sessionEnd(40);
assert.equal(told.length, 1);
assert.deepEqual(told[0][1], { outcome: "refused", code: "write_not_authorized", error: "no", response: refusedV1("w1") });
t.send("w2", 100);
t.tick(600);
t.frame(refusedV1("w2"), 700);
t.frame(admitted("w2"), 800);
assert.equal(t.state("w2")?.outcome === "unconfirmed" && t.state("w2")?.recorded, "refused_late");
});
test("tracker: an id-less error refuses the single unanswered write only, counting answered writes out", () => {
const { t } = tracker();
t.send("w1", 0);
t.send("w2", 1);
t.frame(admitted("w1"), 5);
t.frame(refusedV1(null), 6);
assert.equal(t.state("w1")?.outcome, "admitted");
assert.equal(t.state("w2")?.outcome, "refused");
});
test("tracker: an id-less error never reaches a write that already ended unconfirmed", () => {
const { t } = tracker();
t.send("w1", 0);
t.tick(500);
t.frame(errorV0(), 501);
assert.deepEqual(t.state("w1"), { outcome: "unconfirmed", reason: "window_elapsed" });
});
test("tracker: sessionEnd settles unanswered writes only, and an expired window keeps its own reason", () => {
const { t } = tracker();
t.send("w1", 0);
t.send("w2", 0);
t.send("w3", 400);
t.frame(admitted("w2"), 10);
t.sessionEnd(600);
assert.deepEqual(t.state("w1"), { outcome: "unconfirmed", reason: "window_elapsed" });
assert.equal(t.state("w2")?.outcome, "admitted");
assert.deepEqual(t.state("w3"), { outcome: "unconfirmed", reason: "session_ended" });
});
test("tracker: re-sending a known id is the same write and does not restart its window", () => {
const { t } = tracker();
t.send("w1", 0);
t.send("w1", 400);
t.tick(500);
assert.equal(t.state("w1")?.outcome, "unconfirmed");
});
test("tracker: a send past another write's window settles that write at once", () => {
const { t, told } = tracker();
t.send("w1", 0);
t.send("w2", 600);
assert.deepEqual(told, [["w1", { outcome: "unconfirmed", reason: "window_elapsed" }]]);
});
test("tracker: only receipt and error frames answer; any other frame naming the id settles nothing", () => {
const { t } = tracker();
t.send("w1", 0);
t.frame({ schemaVersion: "cultnet.hello.v0", runtimeId: "r", runtimeKind: "k" }, 1);
t.frame({ schemaVersion: "cultnet.document_put_raw.v0", messageId: "w1" } as unknown as CultNetMessage, 2);
assert.equal(t.state("w1")?.outcome, "pending");
});
test("tracker: nextDeadlineMs reports the nearest unanswered window, none when all settled", () => {
const { t } = tracker();
assert.equal(t.nextDeadlineMs(0), undefined);
t.send("w1", 0);
t.send("w2", 100);
assert.equal(t.nextDeadlineMs(50), 450);
assert.equal(t.nextDeadlineMs(900), 0);
t.frame(admitted("w1"), 60);
assert.equal(t.nextDeadlineMs(150), 450);
t.frame(admitted("w2"), 160);
assert.equal(t.nextDeadlineMs(170), undefined);
});
test("the replay runner passes every vector against this runtime", () => {
const script = join(__dirname, "..", "..", "tools", "replay-write-receipt-vectors.mjs");
const run = spawnSync(process.execPath, [script], { encoding: "utf8" });
assert.equal(run.status, 0, run.stdout + run.stderr);
assert.match(run.stdout, /^PASS scenario\/receipt_admits$/m);
assert.doesNotMatch(run.stdout, /^FAIL/m);
});
test("a failed local send settles only that write unconfirmed as local_send_failed, and an answer after it is recorded late", () => {
const settled: [string, WriteSettlement][] = [];
const tracker = new WriteTracker({ windowMs: 1000, onSettle: (id, settlement) => settled.push([id, settlement]) });
tracker.send("w1", 0);
tracker.send("w2", 0);
tracker.sendFailed("w1", 10);
assert.deepEqual(settled, [["w1", { outcome: "unconfirmed", reason: "local_send_failed" }]]);
assert.equal(tracker.state("w2")?.outcome, "pending");
tracker.frame(admitted("w1"), 20);
assert.deepEqual(tracker.state("w1"), { outcome: "unconfirmed", reason: "local_send_failed", recorded: "admitted_late" });
tracker.sendFailed("w2", 30);
tracker.sendFailed("w2", 31);
assert.equal(settled.length, 2, "a write settles once");
tracker.sendFailed("unknown", 40);
assert.equal(tracker.state("unknown"), undefined);
});
test("a failed local send advances the clock: other writes whose windows have run settle window_elapsed", () => {
const tracker = new WriteTracker({ windowMs: 1000 });
tracker.send("w1", 0);
tracker.send("w2", 500);
tracker.sendFailed("w2", 1200);
assert.deepEqual(tracker.state("w1"), { outcome: "unconfirmed", reason: "window_elapsed" });
assert.deepEqual(tracker.state("w2"), { outcome: "unconfirmed", reason: "local_send_failed" });
});
test("a write whose own send failed settles local_send_failed even when its window has also run", () => {
const tracker = new WriteTracker({ windowMs: 1000 });
tracker.send("w1", 0);
tracker.sendFailed("w1", 1000);
assert.deepEqual(tracker.state("w1"), { outcome: "unconfirmed", reason: "local_send_failed" });
});

View file

@ -0,0 +1,116 @@
#!/usr/bin/env node
// Replays contracts/cultnet/interop/write-receipt-vectors.json against this runtime: every `scenarios.cases` event
// replay through a fake server and the WriteTracker, and every `wire` / `schemaInvalid` frame through the real
// decoder. Prints PASS/FAIL per case; exit 1 on any failure. Runs against the built package (`npm run build` first).
//
// node tools/replay-write-receipt-vectors.mjs [path/to/write-receipt-vectors.json]
//
// The runner interface a port (Rust, C#) must reproduce is documented in README.md, "Write receipt replay runner".
import { readFileSync } from "node:fs";
import path from "node:path";
import { fileURLToPath } from "node:url";
import { createRequire } from "node:module";
const require = createRequire(import.meta.url);
const here = path.dirname(fileURLToPath(import.meta.url));
const { decode, encode } = require("@msgpack/msgpack");
const {
CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS, LEGACY_RESULT_WINDOW_MS, RECEIPT_RESULT_WINDOW_MS, WriteTracker,
encodeCultNetMessageForWire, helloAdvertisesWriteReceipts, parseCultNetMessage, resultWindowMs,
} = require(path.join(here, "..", "dist", "index.js"));
const vectorsPath = process.argv[2] ?? path.join(here, "..", "..", "..", "contracts", "cultnet", "interop", "write-receipt-vectors.json");
const vectors = JSON.parse(readFileSync(vectorsPath, "utf8"));
const results = [];
const check = (name, run) => {
try {
const problems = run();
results.push({ name, problems });
} catch (error) {
results.push({ name, problems: [`threw: ${error instanceof Error ? error.message : String(error)}`] });
}
};
const bytesOf = (hex) => Uint8Array.from(Buffer.from(hex, "hex"));
const hexOf = (bytes) => Buffer.from(bytes).toString("hex").toUpperCase();
// A frame reaches the client as bytes: the fake server encodes it, the client decodes and validates it.
const delivered = (frame) => parseCultNetMessage(decode(encode(frame)));
check("constants", () => [
...(vectors.constants.RECEIPT_RESULT_WINDOW_MS === RECEIPT_RESULT_WINDOW_MS ? [] : [`RECEIPT window ${RECEIPT_RESULT_WINDOW_MS} != ${vectors.constants.RECEIPT_RESULT_WINDOW_MS}`]),
...(vectors.constants.LEGACY_RESULT_WINDOW_MS === LEGACY_RESULT_WINDOW_MS ? [] : [`LEGACY window ${LEGACY_RESULT_WINDOW_MS} != ${vectors.constants.LEGACY_RESULT_WINDOW_MS}`]),
]);
for (const vector of vectors.wire) {
check(`wire/${vector.name}`, () => {
const problems = [];
const parsed = parseCultNetMessage(decode(bytesOf(vector.messagePackHex)));
if (JSON.stringify(sorted(parsed)) !== JSON.stringify(sorted(vector.message))) problems.push(`decoded ${JSON.stringify(parsed)} != ${JSON.stringify(vector.message)}`);
const reencoded = hexOf(encode(encodeCultNetMessageForWire(parsed)));
if (reencoded !== vector.messagePackHex.toUpperCase()) problems.push(`re-encoded bytes ${reencoded} != captured ${vector.messagePackHex}`);
return problems;
});
}
for (const vector of vectors.schemaInvalid) {
check(`schemaInvalid/${vector.name}`, () => {
try {
parseCultNetMessage(decode(bytesOf(vector.messagePackHex)));
} catch {
return [];
}
return [`accepted a frame that must be refused (${vector.why})`];
});
}
for (const scenario of vectors.scenarios.cases) {
check(`scenario/${scenario.name}`, () => {
// The fake server states its capability the only way a real one does: in its hello.
const serverHello = { schemaVersion: "cultnet.hello.v0", runtimeId: "fake-server", runtimeKind: "fake",
...(scenario.server === "receipt" ? { supportedMessageVersions: [...CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS] } : {}) };
const windowMs = resultWindowMs(helloAdvertisesWriteReceipts(serverHello));
const problems = [];
if (windowMs !== vectors.constants[scenario.window]) problems.push(`window ${windowMs} != ${scenario.window}`);
const told = new Map(); // what the caller was told, at settle time
const tracker = new WriteTracker({ windowMs, onSettle: (id, settlement) => {
if (told.has(id)) problems.push(`${id} settled twice`);
told.set(id, settlement);
} });
for (const event of [...scenario.events].sort((a, b) => a.atMs - b.atMs)) {
if ("send" in event) tracker.send(event.send, event.atMs);
else if ("frame" in event) tracker.frame(delivered(event.frame), event.atMs);
else if ("tick" in event) tracker.tick(event.atMs);
else if ("sessionEnd" in event) tracker.sessionEnd(event.atMs);
// transportAck: delivery is not an answer; the tracker has no such input and must not need one.
}
for (const [id, want] of Object.entries(scenario.expect)) {
const got = told.get(id);
if (!got) { problems.push(`${id}: never settled, wanted ${want.outcome}`); continue; }
if (got.outcome !== want.outcome) problems.push(`${id}: told ${got.outcome}, wanted ${want.outcome}`);
if (want.code !== undefined && got.code !== want.code) problems.push(`${id}: code ${got.code}, wanted ${want.code}`);
if (want.reason !== undefined && got.reason !== want.reason) problems.push(`${id}: reason ${got.reason}, wanted ${want.reason}`);
const recorded = tracker.state(id)?.recorded;
if (recorded !== want.recorded) problems.push(`${id}: recorded ${recorded}, wanted ${want.recorded}`);
}
for (const id of told.keys()) if (!(id in scenario.expect)) problems.push(`${id}: settled but the scenario expects nothing for it`);
return problems;
});
}
function sorted(value) {
if (Array.isArray(value)) return value.map(sorted);
if (value && typeof value === "object") return Object.fromEntries(Object.keys(value).sort().map((k) => [k, sorted(value[k])]));
return value;
}
let failed = 0;
for (const { name, problems } of results) {
if (problems.length === 0) { console.log(`PASS ${name}`); continue; }
failed++;
console.log(`FAIL ${name}`);
for (const problem of problems) console.log(` ${problem}`);
}
console.log(`${results.length - failed}/${results.length} passed`);
process.exit(failed === 0 ? 0 : 1);