rhdl_ubft/rpc/
reply.rs

1//! Full-response RPC aggregation and server emission (U29–U32).
2//!
3//! Reference: original uBFT 504eba0835e02af6d384afec6c6c7073530fc11f,
4//! rpc/client.hpp:151–176,377–403 and rpc/server.hpp:152–174.
5//! The original counts messages without retaining sender identity, stores them
6//! in an unchecked five-element array, throws on X,Y,Y, and suppresses ID3.
7//! Here each bounded live request retains one immutable vote per configured
8//! authenticated sender. Equal lengths and every declared payload byte must
9//! match; bytes beyond length are ignored on input and zeroed on output.
10//!
11//! TRUST BOUNDARY: `Reply.sender` must be stamped by an authenticated transport
12//! adapter, never copied from an untrusted packet field. `ServerIn.executed`
13//! must come from identity-preserving application execution, not admission or
14//! proposal acceptance. Neither composition is supplied by this module.
15//!
16//! Configuration is accepted once. Reset retires this instance permanently;
17//! creating a fresh state requires an externally established session/drain
18//! fence. Request IDs start at configured `first_request` (including zero),
19//! are allocated consecutively, and never wrap. Retirement is in request
20//! order even if later requests finish first. This is bounded storage, not a
21//! fixed-size wire protocol: P is configurable and lengths may exceed 64 bytes.
22//! Allocation must precede its first reply by at least one edge: a same-edge
23//! reply for a newly allocated ID is discarded against the registered live set.
24//! The server adapter does not establish exactly-once execution; request IDs
25//! arrive from the trusted identity-preserving execution boundary.
26use rhdl::prelude::*;
27
28/// Frozen client/session and external server-ID map (N odd, 1..=15).
29#[derive(Clone, Copy, Debug, PartialEq, Digital)]
30pub struct Config<const N: usize> {
31    /// One-time configuration offer. Subsequent offers are rejected.
32    pub init: bool,
33    /// Nonzero externally provisioned session namespace.
34    pub session: b64,
35    /// External client identity; zero is allowed.
36    pub client: b32,
37    /// Distinct external replica identities; zero and three are ordinary IDs.
38    pub members: [b32; N],
39    /// First request ID; zero is allowed.
40    pub first_request: b64,
41}
42impl<const N: usize> Default for Config<N> {
43    fn default() -> Self {
44        Self {
45            init: false,
46            session: b64(0),
47            client: b32(0),
48            members: [b32(0); N],
49            first_request: b64(0),
50        }
51    }
52}
53
54/// A response with the namespace and authenticated transport source attached.
55#[derive(Clone, Copy, Debug, PartialEq, Digital)]
56pub struct Reply<const P: usize> {
57    /// Retain all fields until ready.
58    pub valid: bool,
59    /// Stamped by authenticated connection, not a packet assertion.
60    pub sender: b32,
61    /// Session namespace.
62    pub session: b64,
63    /// Client namespace.
64    pub client: b32,
65    /// Exact outstanding request.
66    pub request: b64,
67    /// Number of meaningful bytes, at most P.
68    pub len: b32,
69    /// Full bounded response storage. Tail bytes are normalized to zero.
70    pub data: [b8; P],
71}
72impl<const P: usize> Default for Reply<P> {
73    fn default() -> Self {
74        empty_reply::<P>()
75    }
76}
77
78/// Empty response, usable inside RHDL kernels.
79#[kernel]
80pub fn empty_reply<const P: usize>() -> Reply<P> {
81    Reply::<P> {
82        valid: false,
83        sender: b32(0),
84        session: b64(0),
85        client: b32(0),
86        request: b64(0),
87        len: b32(0),
88        data: [b8(0); P],
89    }
90}
91
92/// Allocation of one outstanding request, before replies can count.
93#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
94pub struct Open {
95    /// Offered request allocation.
96    pub valid: bool,
97    /// Must equal frozen session.
98    pub session: b64,
99    /// Must equal frozen client.
100    pub client: b32,
101    /// Must equal the next consecutive ID.
102    pub request: b64,
103}
104
105/// One distinct server's first accepted reply.
106#[derive(Clone, Copy, Debug, PartialEq, Digital)]
107pub struct Vote<const P: usize> {
108    /// Vote allocated.
109    pub present: bool,
110    /// Declared response length.
111    pub len: b32,
112    /// Canonical response bytes.
113    pub data: [b8; P],
114}
115/// Per-request bounded collector entry.
116#[derive(Clone, Copy, Debug, PartialEq, Digital)]
117pub struct Entry<const N: usize, const P: usize> {
118    /// Slot allocated to request.
119    pub active: bool,
120    /// Exact request ID (not a modulo identity).
121    pub request: b64,
122    /// One immutable vote per membership index.
123    pub votes: [Vote<P>; N],
124    /// A matching distinct-sender majority is frozen.
125    pub complete: bool,
126    /// Frozen majority response length.
127    pub len: b32,
128    /// Frozen majority response bytes.
129    pub data: [b8; P],
130}
131/// Empty collector entry.
132#[kernel]
133pub fn empty_entry<const N: usize, const P: usize>() -> Entry<N, P> {
134    Entry::<N, P> {
135        active: false,
136        request: b64(0),
137        votes: [Vote::<P> {
138            present: false,
139            len: b32(0),
140            data: [b8(0); P],
141        }; N],
142        complete: false,
143        len: b32(0),
144        data: [b8(0); P],
145    }
146}
147
148/// Collector registers; W is 1..=8 and N is odd, 1..=15.
149#[derive(Clone, Copy, Debug, PartialEq, Digital)]
150pub struct CollectorState<const N: usize, const W: usize, const P: usize> {
151    /// One-time configuration has completed.
152    pub initialized: bool,
153    /// Invalid initial configuration or explicit reset retires the instance.
154    pub retired: bool,
155    /// Frozen configuration.
156    pub cfg: Config<N>,
157    /// Bounded request entries.
158    pub entries: [Entry<N, P>; W],
159    /// Oldest live entry.
160    pub head: b3,
161    /// Next allocation slot.
162    pub tail: b3,
163    /// Number of live entries, never greater than W.
164    pub used: b4,
165    /// Next consecutive request ID.
166    pub next_request: b64,
167    /// Maximum request ID has been allocated; no further allocation permitted.
168    pub exhausted: bool,
169    /// Wrapping diagnostic count only; never used for safety decisions.
170    pub rejected: b32,
171    /// Wrapping count of repeated same-sender votes, equal or conflicting.
172    pub duplicates: b32,
173}
174impl<const N: usize, const W: usize, const P: usize> Default for CollectorState<N, W, P> {
175    fn default() -> Self {
176        Self {
177            initialized: false,
178            retired: false,
179            cfg: Config::default(),
180            entries: [empty_entry::<N, P>(); W],
181            head: b3(0),
182            tail: b3(0),
183            used: b4(0),
184            next_request: b64(0),
185            exhausted: false,
186            rejected: b32(0),
187            duplicates: b32(0),
188        }
189    }
190}
191/// Collector inputs.
192#[derive(Clone, Copy, Debug, PartialEq, Digital)]
193pub struct CollectorIn<const N: usize, const P: usize> {
194    /// Configuration offer.
195    pub cfg: Config<N>,
196    /// Terminal retirement; fresh state requires an external reset/drain fence.
197    pub reset: bool,
198    /// Exact request allocation.
199    pub open: Open,
200    /// Authenticated transport response.
201    pub reply: Reply<P>,
202    /// Consumer accepts oldest completed response.
203    pub result_ready: bool,
204}
205impl<const N: usize, const P: usize> Default for CollectorIn<N, P> {
206    fn default() -> Self {
207        Self {
208            cfg: Config::default(),
209            reset: false,
210            open: Open::default(),
211            reply: Reply::default(),
212            result_ready: false,
213        }
214    }
215}
216/// Collector outputs.
217#[derive(Clone, Copy, Debug, PartialEq, Digital)]
218pub struct CollectorOut<const P: usize> {
219    /// Allocation input can be consumed; malformed allocations are rejected.
220    pub open_ready: bool,
221    /// Replies can be consumed, including stale/invalid replies to discard.
222    pub reply_ready: bool,
223    /// Oldest completed response, held unchanged until result_ready.
224    /// Sender is zero: this is a majority result, not a single-server reply.
225    pub result: Reply<P>,
226    /// Initial configuration invalid or instance explicitly retired.
227    pub retired: bool,
228    /// Repeated config offer rejected without altering active state.
229    pub config_rejected: bool,
230}
231
232/// One-cycle collector transition; no latency model or cryptographic oracle.
233#[kernel]
234pub fn collect<const N: usize, const W: usize, const P: usize>(
235    i: CollectorIn<N, P>,
236    r: CollectorState<N, W, P>,
237) -> (CollectorOut<P>, CollectorState<N, W, P>) {
238    let mut s = r;
239    let mut o = CollectorOut::<P> {
240        open_ready: false,
241        reply_ready: false,
242        result: empty_reply::<P>(),
243        retired: r.retired,
244        config_rejected: r.initialized && i.cfg.init,
245    };
246    if !r.initialized && !r.retired && i.cfg.init {
247        let mut valid = N >= 1
248            && N <= 15
249            && (N & 1) == 1
250            && W >= 1
251            && W <= 8
252            && P >= 1
253            && P <= 4294967295usize
254            && i.cfg.session != b64(0);
255        for a in 0..N {
256            for b in 0..N {
257                if a != b && i.cfg.members[a] == i.cfg.members[b] {
258                    valid = false;
259                }
260            }
261        }
262        if valid {
263            s.initialized = true;
264            s.cfg = i.cfg;
265            s.next_request = i.cfg.first_request;
266        } else {
267            s.retired = true;
268        }
269    }
270    let run = r.initialized && !r.retired && !i.reset;
271    if run {
272        o.open_ready = r.used < b4(W as u128) && !r.exhausted;
273        o.reply_ready = true;
274        for k in 0..W {
275            let e = r.entries[k];
276            if r.head == b3(k as u128) && e.active && e.complete {
277                o.result = Reply::<P> {
278                    valid: true,
279                    sender: b32(0),
280                    session: r.cfg.session,
281                    client: r.cfg.client,
282                    request: e.request,
283                    len: e.len,
284                    data: e.data,
285                };
286            }
287        }
288        let retire = o.result.valid && i.result_ready;
289        let allocate = i.open.valid
290            && o.open_ready
291            && i.open.session == r.cfg.session
292            && i.open.client == r.cfg.client
293            && i.open.request == r.next_request;
294        if i.open.valid && o.open_ready && !allocate {
295            s.rejected = s.rejected + b32(1);
296        }
297        if allocate {
298            for k in 0..W {
299                if r.tail == b3(k as u128) {
300                    s.entries[k] = empty_entry::<N, P>();
301                    s.entries[k].active = true;
302                    s.entries[k].request = i.open.request;
303                }
304            }
305            s.tail = if r.tail == b3((W - 1) as u128) {
306                b3(0)
307            } else {
308                r.tail + b3(1)
309            };
310            if r.next_request == !b64(0) {
311                s.exhausted = true;
312            } else {
313                s.next_request = r.next_request + b64(1);
314            }
315        }
316        if retire {
317            for k in 0..W {
318                if r.head == b3(k as u128) {
319                    s.entries[k] = empty_entry::<N, P>();
320                }
321            }
322            s.head = if r.head == b3((W - 1) as u128) {
323                b3(0)
324            } else {
325                r.head + b3(1)
326            };
327        }
328        if allocate && !retire {
329            s.used = r.used + b4(1);
330        }
331        if retire && !allocate {
332            s.used = r.used - b4(1);
333        }
334        if i.reply.valid {
335            let mut member = false;
336            for a in 0..N {
337                if r.cfg.members[a] == i.reply.sender {
338                    member = true;
339                }
340            }
341            let names = i.reply.session == r.cfg.session
342                && i.reply.client == r.cfg.client
343                && i.reply.len <= b32(P as u128)
344                && member;
345            let mut found = false;
346            let mut bytes = [b8(0); P];
347            for b in 0..P {
348                if b32(b as u128) < i.reply.len {
349                    bytes[b] = i.reply.data[b];
350                }
351            }
352            for k in 0..W {
353                let e = r.entries[k];
354                if names
355                    && e.active
356                    && e.request == i.reply.request
357                    && !(retire && r.head == b3(k as u128))
358                {
359                    found = true;
360                    if !e.complete {
361                        for a in 0..N {
362                            if r.cfg.members[a] == i.reply.sender {
363                                if e.votes[a].present {
364                                    s.duplicates = s.duplicates + b32(1);
365                                } else {
366                                    s.entries[k].votes[a] = Vote::<P> {
367                                        present: true,
368                                        len: i.reply.len,
369                                        data: bytes,
370                                    };
371                                }
372                            }
373                        }
374                        // Count matches to each immutable candidate. An outlier
375                        // cannot prevent an honest matching majority later.
376                        for a in 0..N {
377                            let candidate = s.entries[k].votes[a];
378                            let mut count = b5(0);
379                            for b in 0..N {
380                                let v = s.entries[k].votes[b];
381                                if candidate.present
382                                    && v.present
383                                    && candidate.len == v.len
384                                    && candidate.data == v.data
385                                {
386                                    count = count + b5(1);
387                                }
388                            }
389                            if count > b5((N >> 1) as u128) {
390                                s.entries[k].complete = true;
391                                s.entries[k].len = candidate.len;
392                                s.entries[k].data = candidate.data;
393                            }
394                        }
395                    }
396                }
397            }
398            if !found {
399                s.rejected = s.rejected + b32(1);
400            }
401        }
402    }
403    if i.reset {
404        s.retired = true;
405    }
406    (o, s)
407}
408
409/// Execution-to-transport reply adapter configuration.
410#[derive(Clone, Copy, Debug, PartialEq, Digital)]
411pub struct ServerIn<const N: usize, const P: usize> {
412    /// One-time configuration, same namespace as collector.
413    pub cfg: Config<N>,
414    /// Configured external identity, sampled only with initial configuration.
415    pub identity: b32,
416    /// Completed execution; sender field ignored and replaced with identity.
417    pub executed: Reply<P>,
418    /// Transport takes held reply.
419    pub reply_ready: bool,
420    /// Retire this adapter permanently.
421    pub reset: bool,
422}
423impl<const N: usize, const P: usize> Default for ServerIn<N, P> {
424    fn default() -> Self {
425        Self {
426            cfg: Config::default(),
427            identity: b32(0),
428            executed: Reply::default(),
429            reply_ready: false,
430            reset: false,
431        }
432    }
433}
434/// Server adapter registers.
435#[derive(Clone, Copy, Debug, PartialEq, Digital)]
436pub struct ServerState<const N: usize, const P: usize> {
437    /// Configuration accepted.
438    pub initialized: bool,
439    /// Invalid initial configuration or explicit reset.
440    pub retired: bool,
441    /// Frozen session/client/membership.
442    pub cfg: Config<N>,
443    /// Frozen external replica identity.
444    pub identity: b32,
445    /// Held outbound response.
446    pub held: Reply<P>,
447    /// Wrapping diagnostic counter for invalid executions.
448    pub rejected: b32,
449}
450impl<const N: usize, const P: usize> Default for ServerState<N, P> {
451    fn default() -> Self {
452        Self {
453            initialized: false,
454            retired: false,
455            cfg: Config::default(),
456            identity: b32(0),
457            held: Reply::default(),
458            rejected: b32(0),
459        }
460    }
461}
462/// Server adapter outputs.
463#[derive(Clone, Copy, Debug, PartialEq, Digital)]
464pub struct ServerOut<const P: usize> {
465    /// Space for one execution result; invalid namespace/length is discarded.
466    pub execution_ready: bool,
467    /// Held transport reply from our configured identity.
468    pub reply: Reply<P>,
469    /// This instance is terminal.
470    pub retired: bool,
471    /// Configuration offers after initialization are rejected.
472    pub config_rejected: bool,
473}
474/// Every configured replica, including external ID3, emits executed replies.
475#[kernel]
476pub fn serve<const N: usize, const P: usize>(
477    i: ServerIn<N, P>,
478    r: ServerState<N, P>,
479) -> (ServerOut<P>, ServerState<N, P>) {
480    let mut s = r;
481    let mut o = ServerOut::<P> {
482        execution_ready: false,
483        reply: empty_reply::<P>(),
484        retired: r.retired,
485        config_rejected: r.initialized && i.cfg.init,
486    };
487    if !r.initialized && !r.retired && i.cfg.init {
488        let mut own = false;
489        let mut valid = N >= 1
490            && N <= 15
491            && (N & 1) == 1
492            && P >= 1
493            && P <= 4294967295usize
494            && i.cfg.session != b64(0);
495        for a in 0..N {
496            if i.cfg.members[a] == i.identity {
497                own = true;
498            }
499            for b in 0..N {
500                if a != b && i.cfg.members[a] == i.cfg.members[b] {
501                    valid = false;
502                }
503            }
504        }
505        if valid && own {
506            s.initialized = true;
507            s.cfg = i.cfg;
508            s.identity = i.identity;
509        } else {
510            s.retired = true;
511        }
512    }
513    if r.initialized && !r.retired && !i.reset {
514        o.reply = r.held;
515        o.execution_ready = !r.held.valid;
516        if r.held.valid && i.reply_ready {
517            s.held = empty_reply::<P>();
518        }
519        if i.executed.valid && o.execution_ready {
520            if i.executed.session == r.cfg.session
521                && i.executed.client == r.cfg.client
522                && i.executed.len <= b32(P as u128)
523            {
524                s.held = i.executed;
525                s.held.sender = r.identity;
526                for b in 0..P {
527                    if b32(b as u128) >= i.executed.len {
528                        s.held.data[b] = b8(0);
529                    }
530                }
531            } else {
532                s.rejected = r.rejected + b32(1);
533            }
534        }
535    }
536    if i.reset {
537        s.retired = true;
538    }
539    (o, s)
540}
rhdl_ubft/rpc/reply.rs

rhdl_ubft/rpc/
reply.rs
