feat(peer): peer delivery and synchrony recovery - #56
Draft
hadronzoo wants to merge 108 commits into
Draft
Conversation
…ontracts Introduces the value types the peer response path is built from, with no async and no sockets: node and request identifiers, the response frame and its protobuf conversions, the delivery dispositions and their gRPC statuses, and a codec that composes an output codec and an error codec over a whole Result. The frame is encoded in two steps, because a protobuf bytes field writes its length before its contents. The payload is serialized into one bounded scratch allocated once per encoder, the complete frame length is checked against the configured ceiling, and only then is anything framed. A response that would exceed the ceiling is refused before a byte is written, and only a staged, checked frame can be written at all. Decoding bounds the whole message before any per-field work, reads every string and identifier into a bounded stack buffer, and makes one right-sized allocation for the payload. proto/peer.proto and a build script compile the wire contract, so protoc is now a build prerequisite and the CI workflows that compile install it. A test decodes the generated descriptor set, so the hand-written encoder and the .proto cannot drift apart. The compile-time format-id builder moves out of the fixed-width codec module into its own file, since the Result codec now composes ids the same way. No behaviour changes and no tests were removed. Cargo.toml declares an empty workspace table so a checkout nested inside another cargo workspace still builds as a standalone crate.
…falsifiable An explicitly empty `subsystem` or `format` decoded as a valid frame, contradicting the decoder's own rule that those fields exclude their protobuf default by construction. In proto3 an absent singular string and an explicitly empty one denote the same value, so a decoder that rejects the omission must reject the empty string too: `decode_text` now reads a zero-length field as absent, exactly as `decode_id` already did, and the existing missing-field error fires unchanged. The encoder is tightened to match, so it cannot produce a frame its own decoder would refuse: `stage` rejects an empty subsystem alongside an over-long one (the widened `EncodeError::UnusableSubsystem` names both ends of the range), and a compile-time assertion makes a codec with an empty `FORMAT_ID` unusable for responses at all. The rest closes gaps where an invariant the code states had no path to a failing test: - No test could reach `FrameDecodeError::Truncated`, so none of the three "bound the claimed length before you read it" sites were proven. One test now cuts a well-formed frame at every offset and requires each length-delimited field to report the cut itself. - The frame-length model never saw a relay node, so both the model's term and `frame_len`'s were dead in the tests. The encode property now generates the relay. - The scratch never grew inside the encode property, so its shrink-back was asserted where it could not be observed. The property now stages an over-cap response and checks the scratch is back at the cap on every response it accepts. - The three `ErrorCategory` wire discriminants were pinned only by round trips, which move both directions together. A frozen test pins each number and the reserved zero. - `select_host`'s documented error path was unreachable from any test, because the signature named `whoami::Error`, which this crate cannot construct. The error is now a type parameter — production callers still infer `whoami::Error` — and the property covers the failing lookup. Two tests are deleted. `a_refused_response_returns_the_scratch_to_the_cap` is subsumed by `steady_state_encodes_never_reallocate`, which now exercises the same grow-then-shrink-back over generated lengths. `only_an_accepted_response_maps_to_ok` is subsumed by `each_disposition_reports_its_documented_status`, an exhaustive table that pins the exact status of every disposition, `Accepted` included.
…sions structural Three follow-ups from verification of the peer foundations. The frame's protobuf field writers were visible to the whole frame module, and their only caller outside the encoder was a test fixture. That made the encoder's central claim — that a frame can only be written from a staged, cap-checked response — true outside the module and false inside it, where two calls could emit a complete frame of any size. They are private now, and the fixture writes its own fields directly on prost, which also stops the decode suite from checking the encoder against itself. The five dead-code suppressions become expectations gated on not(test). Each item is reached only from its own tests until the transport lands, so a plain expect is unfulfilled in the test build and a plain allow would sit there silently forever. Gated this way the suppression holds exactly while the item really is production-dead and turns into a warning the moment a caller appears, so wiring these up forces the attribute's removal. A reviewer asked for an inherent ResultCodec::new taking two codec instances. Rejected, and the reasoning recorded on the type: Codec requires Default and the framework builds every codec through it, so default() already composes any pair and new would have no caller. No behaviour change and no tests removed.
…fuse repeated frame fields The frame encoder documented an invariant that a trait-legal codec breaks. `Codec::serialize` explicitly permits a codec that owns a wire-format buffer to move it into the caller's `Vec` when that `Vec` is empty, and `stage` hands the codec an empty scratch on every call — so such a codec drops the cap-sized scratch and replaces it with the payload's own, which `shrink_to` cannot undo. The doc claimed the scratch entered every response at exactly the cap, and the allocation property only passed because it was instantiated with a codec that appends. The guarantee is now stated as what the encoder controls: it allocates once, at construction, and staging never reserves; a codec that takes the sanctioned move hands its own buffer over and the encoder reuses that one. The residual is named rather than papered over — the frozen `serialize` signature cannot stop a codec that alternates moving and appending from regrowing the buffer it shrank. The property now runs both codec shapes, so neither half of the trait's contract is untested. The decoder refused an unnamed subsystem, a zero category, and a non-canonical version, but silently took the last of a repeated field. For `payload` that cost one sized allocation per occurrence, so a single capped frame could buy millions of allocate-and-discard pairs. Any second occurrence of any field is now refused, naming the field; an empty occurrence counts, since protobuf spells a present field that way. `NodeId` and `RequestId` were public with no caller, which exempted them from the dead-code gate every other item here relies on and would have committed the crate to their shape a release early. Both are crate-internal until a caller lands, and their constructors now carry the same expectation the gate can retire. That left the ids' own invariants unpinned, so two tests now cover them: request ids are UUIDv7 and sort by mint order, node ids are fresh random values. Also records why the disposition-to-status mapping lives with the dispositions rather than with the transport.
The frame encoder's docs promised more than the code can deliver. They
called the scratch "bounded", said the encoder "reuses" a buffer a moving
codec hands over, and claimed "nothing here allocates per response".
All three are false, and the falsifying case is ordinary rather than
adversarial. Staging checks the scratch's *length* against the cap, never
its capacity, so a codec taking `Codec::serialize`'s sanctioned move into
an empty buffer can hand over a `Vec` whose length fits the cap and whose
capacity is far above it — an application payload built with room to
spare. That response is accepted, leaving the encoder holding an over-cap
scratch. The next response's `shrink_to` then really shrinks it, an
allocator call whose result the codec's next move immediately discards.
And because staging clears the scratch every time, a moving codec moves
every time: the buffer it handed over is dropped, never reused.
State instead what the encoder controls. It allocates one scratch big
enough for a payload at the cap and never grows it — every `reserve` is
the codec's own, through the `&mut Vec<u8>` the frozen `serialize`
signature hands it. What the scratch *is* between responses is therefore
the codec's doing, and staging shrinks it back toward the cap before each
one so nothing accumulates; that shrink is the only place the encoder
itself can allocate after construction, and only after a response left
the scratch over the cap. The in-body comment carried the same false
assumption in shorter form ("whatever a response the cap refused grew the
scratch to") and is corrected with it.
The property could not see any of this: it built every payload with
`collect` and `clone`, which size the allocation to the length, so
capacity always equalled length and the move arm only ever exercised half
its shape. Payloads are now handed over with slack past the cap. Proof
the change is load-bearing: injecting a post-serialize reclaim — which
would make the guarantee true — goes red under the new generator and
green under the old one.
No behaviour changes.
A record can now ask for a peer response by carrying four plain string headers: response-version, response-request-id, response-node, and one response-awaited per subsystem it awaits. They parse, while the record is still borrowed, into a crate-internal RequestTag of two fixed-size ids and no strings. The awaited names repeat rather than being comma-separated because a comma is a legal character in a subsystem name, and every singleton must occur exactly once, so no unstated first-wins or last-wins precedence exists. A record whose reserved headers are duplicated, missing, over-bound, or written under an unsupported revision yields no tag and one counted rejection under a fixed label - never a failed event. Which subsystem a consumer answers for is the name it already configures through PROSODY_SUBSYSTEM, so the decode path and the responder cannot disagree about it. Both decode call sites take it as a required argument: the poll loop, and the message loader. The latter is how a deferred message is re-read - the defer store persists offsets alone, never the record - which is what keeps a deferred request's destination alive across the retry. The Kafka loader's fulfill_requests loses its partition and offset parameters; both were already derivable from the message it is handed. Two loader test configurations that spelled out every field now update the shared one. src/loader/kafka/tests.rs moves to src/loader/kafka/tests/mod.rs so the reload test lands in its own file. That file and src/loader/kafka/mod.rs were already over the line cap and remain so, out of scope here. The rejection counter's export is not asserted - it is a global-meter counter, as the cache-fuse and settle-gate counters are. What is asserted is the classification that picks its label, and that the labels are total and distinct. No tests are deleted.
Pipeline mode was the only mode that re-derived the subsystem a consumer answers peer requests for from the raw configuration field instead of `KeyedStateInputs::subsystem`, whose doc states that one source is what keeps the decode path and the responder from disagreeing. It now carries the value on its middleware stack like its three siblings, so the claim holds everywhere it is cited. Both reads resolve the same immutable field, so no behaviour changes. Two bounds that the suite could not previously distinguish are now pinned: - A record whose reserved response headers are unusable must still reload. Only the tag is lost, never the record — the loader round trip now produces a record with a duplicated `response-version` and asserts it reloads with no destination. - A request naming exactly `MAX_AWAITED` subsystems is legal. The header oracle gained an at-the-bound mutation beside the existing one-past case, so tightening the comparison to `>=` is now detectable. The decode module's doc dropped a numbered step list that had drifted out of order with the code it restated, and two stale claims: this module produces a `DecodedMessage`, not a `ConsumerMessage`, and it parses payloads through the configured codec, never JSON.
A consumer configured with a subsystem name longer than 64 bytes built successfully and could never be addressed. A peer request names the subsystems it awaits in a `response-awaited` header, and the header parser refuses any name a response frame could not carry back, so no producer could reach that consumer and no log said why. Worse, the refusal happens before the awaited match, so one such record was refused by every responder that read it. The bound now belongs to the name. `SubsystemName::MAX_BYTES` is enforced in `try_new`, and the frame type, the frame encoder, the frame decoder and the header parser all read it, so `SUBSYSTEM_MAX_BYTES` is gone. Every path that produces a name is bounded by construction rather than by one configuration field carrying a rule. `SubsystemNameError` becomes an enum so the message names the rule the name broke. Three smaller corrections in the same area: - The header parser dispatches through one `match` over the four reserved names instead of an array scan and an if/else chain whose last arm meant `response-node` only by comment. A fifth reserved name was parsed as a second node id. `RESERVED_HEADERS` has no remaining caller and is deleted, and the redundant `saw_reserved` flag goes with it. - The rejection counter says what it measures. A poll, a deferred reload and a state read each decode the same record, so the counter is decodes, not records. - The request revision is compared as exact bytes. `str::parse::<u8>` accepted `01` and `+1`, while both id headers already demand one text form. `prop_subsystem_name_trims_and_rejects_blank` and `blank_subsystem_names_rejected` are renamed and strengthened, not replaced: the property now checks which rule a refusal reports, and the example now covers a name on the bound and one byte past it. Two mutations pin the revision forms that were accepted before.
The bound on a subsystem name was written in four places. Only one of them was the type. The header parser and the frame encoder each restated "not empty, and no longer than MAX_BYTES", and the frame carried a look-alike `Subsystem` alias whose only reason to exist was that a `SubsystemName` was too expensive to hold. Nothing but a comment tied the four together, and the header parser's own doc admitted it duplicated the constructor. `SubsystemName` now stores its bytes inline. It allocates nothing, so the frame header holds the real type and the alias is gone. The encoder needs no check, because a name it is handed is one a decoder accepts by construction. The header parser calls the new `SubsystemName::checked`, which applies the rule to a borrowed name and copies nothing, so the parser stays allocation free on the poll path without owning a second copy of the rule. The frame decoder builds the name where the untrusted bytes enter, and reports the one case its length bound cannot catch: a name of whitespace alone, which is not empty on the wire and is blank after the trim. `From<&SubsystemName> for Arc<str>` promised a shared allocation that inline storage cannot give, so it is deleted along with its one caller's need for it. `StateReaderError::UnknownPublication` now names the subsystem by its own type. The configuration field needs no `#[validate]` rule and does not get one. In a validator-derived config the rule's idiomatic home is the field's type: a value that exists is already checked, so no rule can be forgotten and none can disagree with the wire. `an_unusable_subsystem_is_refused` is deleted. It proved the encoder refuses an unusable name; `prop_subsystem_name_trims_and_bounds` and `subsystem_name_boundaries` prove such a name cannot be built at all, which is the stronger claim now that the constructor is the only way to make one. Two new tests cover what the change adds: a name at the bound is stored inline, and a whitespace-only name on the wire is refused. Both were proved falsifiable.
A consumer answers a request only when its own subsystem name is among the record's `response-awaited` names. Exact equality is what keeps a response inside the subsystem an operator named, but no test could detect a weaker comparison: every generated case drew the responder from the awaited list, so equality already held, and the one not-awaited case used a name that overlaps no awaited name. A prefix, suffix or substring test — either way round — kept the whole suite green while it answered for a subsystem the record never named. One example test now sends a well-formed request whose single awaited name is a near miss of the responder's: it holds the name, the name holds it, or it differs only in case. Each row was proved red once against the weakening it refuses, then the injection was reverted. Also state the subsystem name rule where an operator reads it: the name is trimmed, and a blank name or one longer than 64 bytes now fails at startup. No test was deleted.
…able Two invariants the parser states were untestable, so a weakening of either kept the whole suite green. The dispatch ends with an arm that ignores every header this protocol does not reserve, and a real record always carries some — a trace context, a source system, whatever the producer set. No case put a foreign header beside a reserved one: the only case carrying one cleared the reserved headers first. Replacing that arm with an early return therefore broke every real record and reddened nothing. The producer header now sits in the base header vector every case is built from, so it rides each mutation, the deterministic row and the rotation property. The id gate accepts one text length, which is what rejects the simple, braced and URN spellings of the same UUID and gives one id one text form. Only the simple and truncated forms were fed, so relaxing the gate to a lower bound accepted braced and URN ids undetected. Two mutations cover them, and the three alternate spellings now share one arm, since each is the same id in a form the protocol refuses. The module doc said an unusable header set always yields one counted rejection. A consumer configured to answer for no subsystem never reads the headers, so it counts nothing. The doc now says so. No test was deleted.
…s cache Every prosody process now publishes where it can be reached, so a peer can be addressed by node id rather than by an address that travelled in a message. A registration is one row per live process, written with a lease and rewritten well inside it by a jittered refresher, deleted outright on clean shutdown, and expired by the lease when a process dies. A second key-only table indexes the live members of a consumer group, scoped by the Kafka cluster that gives a group id its meaning and spread over a fixed number of partitions derived from the node id. Both writes are plain unconditional upserts: node ids are minted fresh at startup and never reused, so there is no lease to acquire and nothing to fence, and the tables need no conditional write anywhere. Resolution reads through a capacity-bounded cache whose entries age out on the same lease the writer uses, so a cached address can never outlive the row that justified it and a burst of traffic for one cold node issues a single read. Retention itself is best-effort, and the cache says so: quick_cache can evict an entry it has just admitted into a full cache, so one read per burst is the guarantee, never that a value stays. The runtime that owns process identity mints the node id, registers before it returns, refreshes, and joins its refresher before it deletes - so a refresh cannot land after the delete and resurrect a departed node. It consults no peer feature flag: every process registers, always. The endpoint is still an input supplied by the caller. The listener that owns a real bound port, the group listing query, the choice between a direct and an advertised endpoint, and the ordering of this shutdown against the rest of a process's teardown are all outside this change. Nothing constructs the runtime yet, so the consumer wiring is still owed. No test was deleted.
`PeerRuntime::start` read every configuration field without running the `Validate` rules that guard them. A cache capacity of zero therefore reached `AddressCache::new`, and the declared bound proved nothing. The runtime now validates first and reports a new `Configuration` error, as every other configuration in this crate does at its constructor. The address cache documents that it caches a known-absent node, so a burst for an id the directory does not hold costs one read. No test held that claim: every id the cache suite asked for was registered up front. A new sub-check of the cache property asks three times for an unknown id and counts one read. Three smaller corrections: - `refresh_delay` re-derived the seconds-to-duration conversion that `RegistrationTtl::duration` owns. It now calls it. - `Endpoint` said its port is the port its listener bound. That is true of a direct endpoint and false of an entry point, whose port an operator sets. `TABLE_NODE_DIRECTORY` said every row names a consumer group; the column is optional. - The `finish` test helper took a `Result<bool>` whose `Ok(false)` arm no caller could produce. It takes a `Result<()>` now. The cache suite also reads its lease off the directory it was built from, rather than converting the same constant a second time.
Three configuration fields did not reach what their names promise, and one reached something it never described. `registration_ttl` reached nothing at all. `PeerRuntime::start` took an already-built `NodeDirectory` and read the lease off it, so an operator who set the field changed nothing, and two independent sources of one value could disagree with no check. `start` now takes the store, builds the directory from the configured lease, and paces the refresher and the address cache from that same value. One configured lease governs all three. The field carries the lease as a `RegistrationTtl`, so a value outside the accepted range cannot reach a configuration at all, and the separate duration validator is gone. A configured `advertised_host` also filled `direct`. The two endpoints then held the same value, which destroyed the reason `direct` exists: a neighbour that shares this node's network label dials `direct` to skip the entry point, so an entry point that reaches `direct` sends every same-network response through a shared component it never needed to touch. `direct` is now discovered and never configured, and the entry point reaches `advertised` alone. `advertised_port` set with no host beside it published a discovered host on a port nothing listens on; a cross-field rule now refuses it. Two claims the code did not deliver are now true. `shutdown` moved the refresh handle out of a lock and awaited it outside, so a second caller found nothing to join and could delete while the refresher was still suspended mid-write; a cancelled join detached the task and left the next caller the same opening. The join and the deletes now run under one lock, and the handle is cleared only after the join returns. `register` claimed a reader never finds an index entry whose node row is missing. Both rows carry separate leases stamped in write order, so after a crash the index always outlives the node row by the gap between the two writes. The doc now states the window the write order really buys, and `deregister` mirrors the order instead of manufacturing the dangling entry deliberately. The address cache told two untruths. Its entries are not fixed size, because a host past 63 bytes leaves `Flexstr`'s inline form, so "every string is inline" was not the reason to weigh items by count. Its age bound is one lease from the moment the read is issued, not from the row's remaining life, so a cached address outlives its row by up to a further lease. Both are now stated as they are. Entries hold the registration behind an `Arc`, so a cache hit costs a reference count rather than a clone of every string on the dialing path. Remaining bounds: the address cache capacity gains a maximum, so a typo cannot ask for a heap the process lacks; the label limit drops to 63 bytes, one under the inline limit it exists to respect. Three tests are added: a configured entry point never reaches `direct`; a removed node stops resolving through `PeerRuntime::resolve`, which is the path a dialer takes and the only path that proves expiry end to end; and the lease range holds at both bounds. Deleted tests. `discovery_prefers_configured_then_routed_then_hostname` is removed with its subject `select_host`, which is deleted and not moved, so no surviving test owes its invariant. The "lease below the minimum" and "lease above the maximum" cases leave `configuration_refuses_degenerate_values` because the field no longer holds a `Duration`; `a_lease_exists_only_inside_its_range` proves that bound at the type that owns it, at both ends rather than at one. The diff is net positive by 41 lines. The additions are the three tests, the cross-field rule, and the capacity maximum; the deletions are `select_host` and its property.
The node directory worked, but four statements about it were false. A refresh delay was a third to a half of the lease, while the doc promised that two lost refreshes still heal. They cannot: the third attempt falls at or after the expiry instant in every draw, before a write's own round trip is counted. The delay is now a fifth to a quarter of the lease, so three delays leave a quarter of the lease unspent and the promise holds. A process pays about 8.9 registration writes per lease instead of 4.8 — two unconditional single-partition writes per refresh, per process, and never per message. The refresh property now asserts that guarantee (three delays plus a quarter of the lease fit inside the lease) instead of the two constants it read off, and takes its name from it. Same subject, strictly stronger. The routed-address discovery kept its composition when `select_host` went away, but no test covered either source of the direct host, so the record that no surviving test owed that invariant was wrong. One test now proves both: the routed address while the probe answers, and this machine's name where it finds none. It aims the probe at an address so two probes cannot land on different address families of one name, and reaches the fallback with a contact point that has no port, so it needs no network. The address cache ages entries on this process's own configured lease, not on the lease the writer applies to its row. A process configured with the longer lease keeps an entry after the row is gone. The doc claimed the two were the same lease. Single flight is best-effort as well: a fill that fails inserts nothing, so the next waiter issues its own read. The doc claimed the opposite. `RegistrationTtl::DEFAULT` and `MIN` no longer describe themselves through the old margin. Falsification: the old pacing reds the refresh property; publishing the machine name always reds the first arm of the discovery test; a different fallback reds the second. Each injection was reverted.
"The map cannot grow with traffic" was absolute where the code is not. `quick_cache` holds a miss's placeholder in its entry slab and weighs only resident entries, so a burst of misses on distinct node ids holds placeholders outside the declared capacity until each fill inserts or drops. The bound is still real, because the callers in flight bound the placeholders and the ids an outsider picks do not. The sentence now says which entries the capacity counts and names the placeholder's removal path, and it uses the same vocabulary as the crate's other bounded cache, which already separates declared weight from process RSS. The capacity paragraph and the age paragraph are also separated, because they state two different bounds and the one paragraph had grown past a reader's patience. Doc text only: no type, no statement, and no test changes.
The probe's doc promised its answer was "by construction an address on a network the rest of the deployment already shares". The routing table answers for the Cassandra contact point, and peers are a different set of hosts. A contact point on the loopback interface makes every process publish a loopback address, so every dial reaches the dialer. A host that reaches Cassandra over a management interface publishes the management address, which a peer on the data network may not reach. `direct` is the address a response is sent to, so a guarantee stated about it is load-bearing for a later reader. The doc now says what the probe answers, gives the two cases where the answer serves no peer, and names the configured entry point as the escape hatch. The mechanism, the discovery order, and every statement are unchanged. No test changed and no test was deleted.
…ission gate A response leaves a settled event through an apply hook, and an apply hook is per-key serialized: the next event for the same key waits for it. So the hook must never await the network and never await queue capacity. This adds the delivery path that makes that possible. A process-owned fleet holds a fixed table of live destinations, each with a fixed number of send slots, a rate limit and a monotonic use stamp. Reserving a slot takes one lock, scans a fixed array and takes a semaphore permit — it never awaits. A destination with sends in flight is never evicted, because taking a slot happens under the same lock that clears a cell. When every cell is busy the new destination is refused and counted. The table's length and its slot total are validated at startup against a ceiling, and so is the encode buffer the frame ceiling implies. Admission runs through a gate with a count rather than a flag, so shutdown can close it and wait for every hook that already entered to leave. The closed bit and the count share one word, which makes "closed between the check and the reservation" unrepresentable. A reservation hands its slot on through a closure and leaves the gate only once that returns, so a drained gate also means nobody is still about to queue work. The typed half sits beside the fleet rather than inside it, because a queued response is a moved handler result and one process may run consumers with different result types. It owns one worker per destination, and each worker owns one codec instance and one encode buffer sized once to the frame ceiling. The worker paces, resolves the address, encodes into its buffer and delivers under one deadline, trying again only for the failures whose outcome another attempt could resolve. Pacing comes before the encode, so a response that spends its deadline waiting is never encoded at all. Delivery reaches the network through one seam, so the responder knows nothing about the transport. This change adds the in-process implementation the tests drive it through. The frame's own writer moves behind that seam, so a staged response is delivered as bytes and the router keeps no response vocabulary.
A response paced past its deadline must be dropped before it is encoded: the pacing wait comes first so that a response which spends its deadline waiting costs no work at all. The test that names that invariant asserted only what reached the transport, what the drop counter held, and that the slot went back. None of those observe an encode. That gap is provable rather than theoretical. Move the pacing wait after the encode, so the order becomes address, encode, pace, deliver, and the invariant is false — yet every sender test still passes. The obstacle was observability. A delivery worker builds its own codec through `Default`, so a suite holds no handle on the instance that encodes and cannot read that instance's counter. The test codec now also counts into a thread-local total. The sender suites run one current-thread runtime, so the worker encodes on the thread that drives it, and the test reads the total as a difference across the run rather than as an absolute. A thread-local rather than a process-wide static, because a static would be correct only while tests never share a process — a property of the runner, not of the test.
…e the encode Two defects made a bound depend on a caller remembering it. `FleetConfiguration::validate()` had no production caller, so a zero rate, a zero deadline, a table of no cells and an over-ceiling slot product all reached live code. `DestinationFleet::new` now validates and returns a `Result`, so an unvalidated fleet does not exist. The degenerate-field test asserts through that constructor rather than through the derive it used to call directly. `timeout_at` polls its inner future before it consults the delay, so a job whose deadline had already passed was paced, encoded and handed to the transport before the timeout could fire. A worker held past a deadline by the job ahead of it reached exactly that. The pipeline now runs under a biased `select!` whose first arm is the deadline, so a passed deadline wins before anything is polled. `deliver_job` returns `bool` and its caller counts the outcome, which replaces six scattered counter writes with one per dequeued job. Sent plus dropped now equals queued by construction, and the sender's harness asserts it on drain. The remaining changes make claims resolve and give the phase's invariants tests that can go red: - A reservation leaves the admission gate only after its slot is handed on. Nothing tested this, and it is what a shutdown drain depends on. - Pacing never hands out a turn in the past. Every earlier test started from an unpaced limiter, so the anti-burst clause was unfalsified. - A node never occupies two cells, a live node with no free slot is refused for its slots, and a retry that succeeds stops. - A destination's stamp is now taken before the permit attempt. A saturated destination never refreshed it, so it became the first eviction candidate the moment it went idle, losing its cell under the load that wanted it kept. - Doc claims that were false are corrected: the fleet's "only allocation", the scratch budget's scope (now per sender, and the constant is renamed to say so), the direction the rate limit bounds, and the lane allocation on the hook path. - `AddressResolver` now only reads, and only through its cache. The runtime holds its own directory handle for the writes it makes about itself. Cassandra was unreachable in this environment, so the router suites that need it could not run. They fail identically on tests this change does not touch.
hadronzoo
changed the title
A response used to build its destination's queue and start its worker on the hook path, whenever the fleet cell held no queue or held one for an earlier occupant. Past the configured table size that is every response: a channel, a task, and a cap-sized encode buffer each time. The queues and the workers are now built once, one per cell, and a job carries the destination it is paced against. The queue store needs no lock and no generation, and a sender can be drained: it drops the queues and joins the workers, bounded by the send deadline each response already carries. The table is now an array of cells rather than an array of pointers. Both scans a reservation runs — find this node, choose an eviction — read the node id and the use stamp from the cell itself, and reach a destination record only for the cell they select. An encoder now gives its scratch back when its response is finished rather than when the next one starts, so a worker waiting on a quiet destination holds none of the last response's bytes. Also: a sender holds the fleet it sized itself from, so the cell a reservation names always indexes its queues; `Reservation::commit` reports only whether the slot was taken, so the slot cannot leave through the return value; and the startup ceiling is asserted through the constructor that enforces it. Corrects four recorded justifications that did not describe the code: what a slot count bounds, what a destination's slots and rate limit bound, what an encode buffer costs, and what a destination's pacing survives. New: a response-layer property over a stream of responses naming more nodes than the table holds — the fleet stays bounded, every refusal is counted, every accepted response is accounted for, and each one reaches the node it was queued for; a deterministic eviction example through the sender; and a pin on what a released encoder holds.
Two senders may share one fleet, so a destination can see one send in flight from each of them. Three docs claimed a destination is sent to one response at a time. Scope that claim to the sender that makes it. Admitting a destination allocates one record, which the fleet's own constructor documents. The sender's docs claimed the send path allocates nothing at all. Say what a send really reserves: no buffer, no queue and no task. Build each worker's encoder in `TypedSender::new` and move it into the task, so "everything a send needs exists before the first one" is a fact of the code rather than a hope about when a spawned task first runs. Correct four more claims that do not resolve: the fleet counts its own refusals, a live reservation may outlast the start of a close, a host past the inline capacity does allocate, and the drain's deadline is measured between polls. `a_queued_response_survives_the_sender_that_queued_it` drained its sender, which joins the workers, so nothing risked the survival its name states. It now drops the sender and waits for the delivery. A sender that cancelled its workers on drop reds it, and reds nothing else.
Add the respond layer, applied at the chain terminator so it sits directly around the application handler and nowhere else. It reads the request tag off the message, carries it on both result arms, and - from after_commit only - moves the typed result into a destination slot. A dispatch with no tag, and every after_abort, forward the result to the handler's own hook untouched. A timer dispatch carries no tag by construction, so it cannot answer at all; a deferred reload arrives as a message dispatch and does. The layer never reads an error category to decide whether to send: which apply hook fired decides that, so a transient failure that exhausts its retries answers its requester while the attempts before it stay silent. The category rides the frame as a label only. A response frame now states a typed status, so a success no longer has to claim an error category. The wire keeps the three category discriminants and adds one for a success; zero stays reserved, so an omitted field is still malformed. The landed frozen bytes for an error frame do not change. The typed sender hands a refused response back instead of dropping it, so a result that cannot reserve a slot still reaches the handler's own hook unencoded. The test router that drives delivery moves from the sender's test tree into the loopback module, so the respond suites reuse it rather than copy it. No test was deleted. The frozen-bytes test one_response_frames_to_known_bytes grew a success row rather than gaining a second test beside it.
Two pins in the respond suite stayed green under a defect they were meant to catch. The retry cascade covered only the direction that settles on a success. An error arm that dropped its response metadata therefore lost its answer with no test to say so, because the failed attempt was retried away and never settled. One test now runs both settling directions over one retry ceiling: a first attempt that fails and then succeeds, and a transient error that exhausts its retries. Each direction reads the decoded frame, so a lost answer and a mislabelled one both go red. This replaces `retry_exhaustion_responds_exactly_once`, whose count assertions the new test repeats for the same session, and adds the frame the old test never read. The stall pin could only detect a defect through its deadline, and the one shape of stall it was aimed at is unwritable: a destination's queue is as deep as that destination has slots, so a queue push never waits. It now reads the send counters after both dispatches and states the invariant positively — no response had finished when the hooks returned. A hook that awaits its delivery outcome reds it on that assertion. The doc names the writable shapes of the defect, so the next reader aims at one of them.
A category varint wider than the field's `int32` folded onto a status: a peer that sent 2^32 + 4 read back as `Success`. That is the one field that decides what a requester believes happened to its request. The decoder now narrows through `i32::try_from` and refuses what does not fit, which is the posture it already takes toward a too-wide `protocol_version`. One row in the malformed-frame table pins it. `ErrorCategory`'s wire mapping now records that `4` is reserved for a response frame's success, so a fourth category cannot silently decode as one. One assert pins that a responder never sets the relay. The docs this area carried are corrected in the same pass: * The respond module doc named a symbol that does not exist, and restated the invariant its constructor already states. * The respond layer's bound doc gave the wrong reason for its four bounds. The codec's payload is the handler's own result, and a codec payload must be `Send + Sync + 'static`. * `Responder::drain` did not say that it can start only after the last handler is dropped. * The `RequestTag::header` dead-code reason named an exerciser that does not name the method. * `SendCounters::dropped` said a fleet refusal never reaches a sender, which it now does, and did not say that a queue refusal hands its payload back. * The delivery suites' module doc described two of its four suites. Two cleanups finish work this area started: the `RequestId` impls sit beside the type they belong to, and the fleet suites reuse the loopback helpers rather than keep byte-identical copies. The frozen-bytes test is renamed for the two rows it now freezes.
…ire too A response frame states a typed status: the three error categories, plus one discriminant for a success. The wire vocabulary still said category. Rename the `.proto` field, the field constant, the decode strings and the two decode error variants to status, so one concept has one name from the schema to the error message. Nothing on the wire changes. Field 6 keeps its number, its `int32` type and its discriminants, a proto3 field name never reaches the wire, and the frozen bytes are untouched. The proto-parity test holds the schema and the hand-written codec to the new name. The responder's doc stated a bare obligation, that its subsystem must match the name that admitted the request tag. It now names the one source both reads, `KeyedStateInputs::subsystem`, so the wiring cannot satisfy the obligation from memory. No test is deleted here. An earlier commit in this work deleted `retried_attempts_stay_silent` without naming its survivor: both of its invariants live in `a_retried_cascade_answers_once_with_the_settled_outcome`, which runs both settling directions and reads the decoded frame the old test never read.
The typed sender returned a refusal class beside the payload, and no production caller ever read it. The respond hook forwards the result whatever the class, and each class is already counted where it happens: the fleet counts a refusal of a slot, the sender counts a job its queue could not take. So `send` now returns the payload alone, and `Rejected` and `Refused` are gone. The bounds property loses nothing. It counted fleet refusals and held them to the fleet's own counter; it now counts every refusal against that same counter, so a queue refusal - the case the deleted arm named - still reds the assertion. The respond layer's abort arm now takes the carriers apart itself and never binds the request tag, so it holds nothing a frame header could be built from. That leaves the commit arm as the only place a tag is reachable, and `split` has no second caller to justify it. Two doc corrections: the module's own tests build the layer directly, so `responding_provider` is the only way to build it from outside the module; and `RespondError` records why it sits beside `Responded` instead of at the end of the file.
The Cassandra model property retains the directory model invariant. The Cassandra label, deregistration, and lease tests retain their matching invariants. The cache property retains in-process directory coverage. The removed capacity test covered only the deleted implementation.
The exhaustive wire test replaces and strengthens the randomized wire disposition property and its separate coverage test.
a_drop_names_its_reason_and_never_the_node now subsumes an_unpublished_node_is_never_dialed. It also proves no dial occurs and the slot returns.
the_listener_serves_the_caps_it_was_configured_with subsumes a_connection_over_the_cap_is_refused_and_counted through the real listener.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Draft.
Why
Kafka publishing is one-way. A producer learns that Kafka accepted a message, but it does not learn the handler result.
Peer delivery lets a caller request results from named consumer subsystems. Each response starts only after the responder commits and publishes its durable work.
The caller receives one flat outcome for each requested subsystem. Missing responses become timeout outcomes. A response remains best-effort and is not a delivery guarantee.
What changed
HighLevelClientandProsodyRequester.The full design is in
docs/peer/design.md.Security deferral
The peer listener uses plaintext transport and does not authenticate peers. Deploy it only on a trusted network or behind a trusted service mesh.
Add transport security and peer authorization before you expose the listener to an untrusted network.
src/router/grpc/mod.rs::serveis the transport attachment point.