rhdl_ubft/rpc/
execution.rs

1//! Ordered RPC execution/cache adapter; see docs/rpc-execution-controller-contract.md.
2//!
3//! TRUST: decided kind/session/key and application bytes come from immutable,
4//! authenticated admission plus ordered consensus. This module does not supply
5//! that boundary, checkpoint restore, or authentication. ACKs are separate
6//! decided control operations. Never use a transport acceptance as an ACK.
7//! One application operation is active; client sequence order may have holes.
8//! Reset/fence is terminal; presented tokens and uncertain effects cannot retry.
9use rhdl::prelude::*;
10
11#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
12pub struct Key {
13    pub client: b32,
14    pub session: b64,
15    pub sequence: b64,
16    pub hash: [b64; 4],
17}
18#[derive(Clone, Copy, Debug, PartialEq, Digital)]
19pub struct Config<const C: usize> {
20    pub valid: bool,
21    pub clients: [b32; C],
22    pub sessions: [b64; C],
23}
24#[derive(Clone, Copy, Debug, PartialEq, Digital)]
25pub struct Decided<const P: usize> {
26    pub valid: bool,
27    /// Authenticated domain: 1 REQUEST, 2 cumulative ACK; not packet metadata.
28    pub kind: b8,
29    pub key: Key,
30    pub len: b32,
31    pub data: [b8; P],
32}
33#[derive(Clone, Copy, Debug, PartialEq, Digital)]
34pub struct Dispatch<const P: usize> {
35    pub valid: bool,
36    pub token: b64,
37    pub key: Key,
38    pub len: b32,
39    pub data: [b8; P],
40}
41#[derive(Clone, Copy, Debug, PartialEq, Digital)]
42pub struct Completion<const R: usize> {
43    pub valid: bool,
44    pub token: b64,
45    pub key: Key,
46    /// 0 committed success, 1 committed application error, 2 uncertain, 3 invalid.
47    pub outcome: b2,
48    pub status: b32,
49    pub len: b32,
50    pub data: [b8; R],
51}
52#[derive(Clone, Copy, Debug, PartialEq, Digital)]
53pub struct Reply<const R: usize> {
54    pub valid: bool,
55    pub key: Key,
56    pub outcome: b2,
57    pub status: b32,
58    pub len: b32,
59    pub data: [b8; R],
60}
61#[derive(Clone, Copy, Debug, PartialEq, Digital, Default)]
62pub struct Notice {
63    pub valid: bool,
64    pub kind: b8,
65    pub key: Key,
66    /// 0 ACK applied,1 ACK no-op,2 ACK ahead,3 namespace,4 window,
67    /// 5 retired,6 conflict,7 malformed,8 config rejected.
68    pub status: b4,
69    pub state_valid: bool,
70    pub acked_through: b64,
71    pub completed_through: b64,
72}
73#[derive(Clone, Copy, Debug, PartialEq, Digital)]
74pub struct Entry<const P: usize, const R: usize> {
75    /// 0 empty,1 pending,2 executing,3 completed.
76    pub phase: b2,
77    pub key: Key,
78    pub len: b32,
79    pub data: [b8; P],
80    pub token: b64,
81    pub result: Reply<R>,
82    pub replay: bool,
83}
84#[derive(Clone, Copy, Debug, PartialEq, Digital)]
85pub struct State<const C: usize, const W: usize, const P: usize, const R: usize> {
86    pub initialized: bool,
87    pub retired: bool,
88    pub clients: [b32; C],
89    pub sessions: [b64; C],
90    pub acked: [b64; C],
91    pub completed: [b64; C],
92    /// Reserved W slots per client: another client cannot consume its capacity.
93    pub entries: [[Entry<P, R>; W]; C],
94    pub next_token: b64,
95    pub token_exhausted: bool,
96    pub busy: bool,
97    pub accepted: bool,
98    /// Full dispatch retained after acceptance for completion correlation.
99    pub work: Dispatch<P>,
100    /// Detached copy: ACK/reuse cannot modify a stalled output.
101    pub reply: Reply<R>,
102    /// Next circular cache entry to consider after installing a held reply.
103    pub reply_client: b3,
104    pub reply_slot: b3,
105    pub notice: Notice,
106    pub terminal_completion: Completion<R>,
107    pub stale_completions: b32,
108}
109#[derive(Clone, Copy, Debug, PartialEq, Digital)]
110pub struct Input<const C: usize, const P: usize, const R: usize> {
111    pub cfg: Config<C>,
112    pub decided: Decided<P>,
113    pub app_ready: bool,
114    pub completion: Completion<R>,
115    pub reply_ready: bool,
116    pub notice_ready: bool,
117    pub fence: bool,
118}
119#[derive(Clone, Copy, Debug, PartialEq, Digital)]
120pub struct Output<const P: usize, const R: usize> {
121    pub decided_ready: bool,
122    pub dispatch: Dispatch<P>,
123    pub completion_ready: bool,
124    pub reply: Reply<R>,
125    pub notice: Notice,
126    pub retired: bool,
127}
128
129#[kernel]
130pub fn empty_key() -> Key {
131    Key {
132        client: b32(0),
133        session: b64(0),
134        sequence: b64(0),
135        hash: [b64(0); 4],
136    }
137}
138#[kernel]
139pub fn empty_reply<const R: usize>() -> Reply<R> {
140    Reply::<R> {
141        valid: false,
142        key: empty_key(),
143        outcome: b2(0),
144        status: b32(0),
145        len: b32(0),
146        data: [b8(0); R],
147    }
148}
149#[kernel]
150pub fn empty_completion<const R: usize>() -> Completion<R> {
151    Completion::<R> {
152        valid: false,
153        token: b64(0),
154        key: empty_key(),
155        outcome: b2(0),
156        status: b32(0),
157        len: b32(0),
158        data: [b8(0); R],
159    }
160}
161#[kernel]
162pub fn empty_entry<const P: usize, const R: usize>() -> Entry<P, R> {
163    Entry::<P, R> {
164        phase: b2(0),
165        key: empty_key(),
166        len: b32(0),
167        data: [b8(0); P],
168        token: b64(0),
169        result: empty_reply::<R>(),
170        replay: false,
171    }
172}
173impl<const C: usize> Default for Config<C> {
174    fn default() -> Self {
175        Self {
176            valid: false,
177            clients: [b32(0); C],
178            sessions: [b64(0); C],
179        }
180    }
181}
182impl<const P: usize> Default for Decided<P> {
183    fn default() -> Self {
184        Self {
185            valid: false,
186            kind: b8(0),
187            key: empty_key(),
188            len: b32(0),
189            data: [b8(0); P],
190        }
191    }
192}
193impl<const P: usize> Default for Dispatch<P> {
194    fn default() -> Self {
195        Self {
196            valid: false,
197            token: b64(0),
198            key: empty_key(),
199            len: b32(0),
200            data: [b8(0); P],
201        }
202    }
203}
204impl<const R: usize> Default for Completion<R> {
205    fn default() -> Self {
206        empty_completion::<R>()
207    }
208}
209impl<const R: usize> Default for Reply<R> {
210    fn default() -> Self {
211        empty_reply::<R>()
212    }
213}
214impl<const C: usize, const W: usize, const P: usize, const R: usize> Default for State<C, W, P, R> {
215    fn default() -> Self {
216        Self {
217            initialized: false,
218            retired: false,
219            clients: [b32(0); C],
220            sessions: [b64(0); C],
221            acked: [b64(0); C],
222            completed: [b64(0); C],
223            entries: [[empty_entry::<P, R>(); W]; C],
224            next_token: b64(1),
225            token_exhausted: false,
226            busy: false,
227            accepted: false,
228            work: Dispatch::default(),
229            reply: Reply::default(),
230            reply_client: b3(0),
231            reply_slot: b3(0),
232            notice: Notice::default(),
233            terminal_completion: Completion::default(),
234            stale_completions: b32(0),
235        }
236    }
237}
238impl<const C: usize, const P: usize, const R: usize> Default for Input<C, P, R> {
239    fn default() -> Self {
240        Self {
241            cfg: Config::default(),
242            decided: Decided::default(),
243            app_ready: false,
244            completion: Completion::default(),
245            reply_ready: false,
246            notice_ready: false,
247            fence: false,
248        }
249    }
250}
251
252/// Prestate outputs transfer this edge; next-state completion advances the
253/// watermark, but a decided ACK cannot transfer on that completion edge.
254#[kernel]
255pub fn step<const C: usize, const W: usize, const P: usize, const R: usize>(
256    i: Input<C, P, R>,
257    r: State<C, W, P, R>,
258) -> (Output<P, R>, State<C, W, P, R>) {
259    let mut s = r;
260    let mut o = Output::<P, R> {
261        decided_ready: false,
262        dispatch: r.work,
263        completion_ready: r.initialized,
264        reply: r.reply,
265        notice: r.notice,
266        retired: r.retired || i.fence,
267    };
268    if r.reply.valid && i.reply_ready {
269        s.reply.valid = false;
270    }
271    if r.notice.valid && i.notice_ready {
272        s.notice.valid = false;
273    }
274    // Already-visible valid may not be retracted, even by a terminal fence.
275    if r.work.valid && i.app_ready {
276        s.work.valid = false;
277        s.accepted = true;
278        for c in 0..C {
279            for w in 0..W {
280                if r.entries[c][w].phase == b2(1) && r.entries[c][w].token == r.work.token {
281                    s.entries[c][w].phase = b2(2);
282                }
283            }
284        }
285    }
286    if i.fence {
287        s.retired = true;
288    }
289    if !r.initialized && !s.retired && i.cfg.valid {
290        // Unsuffixed literals: RHDL preserves token text for its integer parser.
291        // Rust infers usize here and u128 inside b64(...) below.
292        let mut good = C >= 1
293            && C <= 8
294            && W >= 1
295            && W <= 8
296            && P >= 1
297            && P <= 4294967295
298            && R >= 1
299            && R <= 4294967295;
300        for c in 0..C {
301            if i.cfg.sessions[c] == b64(0) {
302                good = false;
303            }
304            for d in 0..C {
305                if c != d && i.cfg.clients[c] == i.cfg.clients[d] {
306                    good = false;
307                }
308            }
309        }
310        if good {
311            s.initialized = true;
312            s.clients = i.cfg.clients;
313            s.sessions = i.cfg.sessions;
314        } else {
315            s.retired = true;
316        }
317    }
318    if r.initialized && i.cfg.valid && !s.retired && !s.notice.valid {
319        s.notice = Notice {
320            valid: true,
321            kind: b8(0),
322            key: empty_key(),
323            status: b4(8),
324            state_valid: false,
325            acked_through: b64(0),
326            completed_through: b64(0),
327        };
328    }
329    if r.initialized && i.completion.valid {
330        let active_token = r.busy && i.completion.token == r.work.token;
331        if s.retired {
332            if active_token && !s.terminal_completion.valid {
333                s.terminal_completion = i.completion;
334            }
335        } else if active_token {
336            let good = r.accepted
337                && i.completion.key == r.work.key
338                && i.completion.len <= b32(R as u128)
339                && i.completion.outcome <= b2(1);
340            if !good {
341                s.retired = true;
342                s.terminal_completion = i.completion;
343            } else {
344                let mut data = [b8(0); R];
345                for b in 0..R {
346                    if b32(b as u128) < i.completion.len {
347                        data[b] = i.completion.data[b];
348                    }
349                }
350                for c in 0..C {
351                    for w in 0..W {
352                        if r.entries[c][w].phase == b2(2) && r.entries[c][w].token == r.work.token {
353                            s.entries[c][w].phase = b2(3);
354                            s.entries[c][w].result = Reply::<R> {
355                                valid: true,
356                                key: r.work.key,
357                                outcome: i.completion.outcome,
358                                status: i.completion.status,
359                                len: i.completion.len,
360                                data,
361                            };
362                            s.entries[c][w].replay = true;
363                        }
364                    }
365                }
366                s.busy = false;
367                s.accepted = false;
368            }
369        } else {
370            s.stale_completions = r.stale_completions + b32(1);
371        }
372    }
373    if r.initialized && !s.retired && i.decided.valid && !s.notice.valid {
374        let d = i.decided;
375        let same_active = r.busy
376            && d.kind == b8(1)
377            && d.key.client == r.work.key.client
378            && d.key.session == r.work.key.session
379            && d.key.sequence == r.work.key.sequence;
380        if !r.busy || same_active {
381            let mut found_client = false;
382            let mut status = b4(3);
383            let mut reject = true;
384            let mut floor = b64(0);
385            let mut completed = b64(0);
386            let mut canonical = [b8(0); P];
387            for b in 0..P {
388                if b32(b as u128) < d.len {
389                    canonical[b] = d.data[b];
390                }
391            }
392            for c in 0..C {
393                if d.key.client == r.clients[c] && d.key.session == r.sessions[c] {
394                    found_client = true;
395                    floor = r.acked[c];
396                    completed = r.completed[c];
397                    if d.kind == b8(2) {
398                        let mut empty = d.len == b32(0);
399                        for b in 0..P {
400                            if d.data[b] != b8(0) {
401                                empty = false;
402                            }
403                        }
404                        if !empty {
405                            status = b4(7);
406                        } else if d.key.sequence <= floor {
407                            status = b4(1);
408                        } else if d.key.sequence > completed {
409                            status = b4(2);
410                        } else {
411                            status = b4(0);
412                            s.acked[c] = d.key.sequence;
413                            floor = d.key.sequence;
414                            for w in 0..W {
415                                if r.entries[c][w].phase == b2(3)
416                                    && r.entries[c][w].key.sequence <= d.key.sequence
417                                {
418                                    s.entries[c][w] = empty_entry::<P, R>();
419                                }
420                            }
421                        }
422                    } else if d.kind != b8(1) || d.len > b32(P as u128) {
423                        status = b4(7);
424                    } else if d.key.sequence <= floor {
425                        status = b4(5);
426                    } else if d.key.sequence - floor > b64(W as u128) {
427                        status = b4(4);
428                    } else {
429                        let mut found = false;
430                        for w in 0..W {
431                            let e = r.entries[c][w];
432                            if e.phase != b2(0) && e.key.sequence == d.key.sequence {
433                                found = true;
434                                if e.key == d.key && e.len == d.len && e.data == canonical {
435                                    reject = false;
436                                    if s.entries[c][w].phase == b2(3) {
437                                        s.entries[c][w].replay = true;
438                                    }
439                                } else {
440                                    status = b4(6);
441                                }
442                            }
443                        }
444                        if !found && !r.busy {
445                            let mut taken = false;
446                            for w in 0..W {
447                                if r.entries[c][w].phase == b2(0) && !taken {
448                                    taken = true;
449                                    reject = false;
450                                    if r.token_exhausted {
451                                        s.retired = true;
452                                    } else {
453                                        s.entries[c][w] = Entry::<P, R> {
454                                            phase: b2(1),
455                                            key: d.key,
456                                            len: d.len,
457                                            data: canonical,
458                                            token: r.next_token,
459                                            result: empty_reply::<R>(),
460                                            replay: false,
461                                        };
462                                        s.work = Dispatch::<P> {
463                                            valid: true,
464                                            token: r.next_token,
465                                            key: d.key,
466                                            len: d.len,
467                                            data: canonical,
468                                        };
469                                        s.busy = true;
470                                        s.accepted = false;
471                                        if r.next_token == b64(18446744073709551615) {
472                                            s.token_exhausted = true;
473                                        } else {
474                                            s.next_token = r.next_token + b64(1);
475                                        }
476                                    }
477                                }
478                            }
479                            // Impossible for a valid unique in-window request with W/client.
480                            // Fail closed on inconsistent/restored state, never bypass the head.
481                            if !taken {
482                                s.retired = true;
483                                reject = false;
484                            }
485                        }
486                    }
487                }
488            }
489            o.decided_ready = !s.retired;
490            if reject && !s.retired {
491                s.notice = Notice {
492                    valid: true,
493                    kind: d.kind,
494                    key: d.key,
495                    status,
496                    state_valid: found_client,
497                    acked_through: floor,
498                    completed_through: completed,
499                };
500            }
501        }
502    }
503    if r.initialized && !s.retired {
504        // Read the post-completion state; ACK above used only registered watermark.
505        for c in 0..C {
506            for pass in 0..W {
507                let _unused = pass;
508                if s.completed[c] != b64(18446744073709551615) {
509                    let next = s.completed[c] + b64(1);
510                    let mut exists = false;
511                    for w in 0..W {
512                        if s.entries[c][w].phase == b2(3) && s.entries[c][w].key.sequence == next {
513                            exists = true;
514                        }
515                    }
516                    if exists {
517                        s.completed[c] = next;
518                    }
519                }
520            }
521        }
522        if !s.reply.valid {
523            let mut copied = false;
524            let mut client = r.reply_client;
525            let mut slot = r.reply_slot;
526            // Visit every reserved entry once, starting after the last copy.
527            // Persistent low-index duplicates cannot monopolize the output.
528            for client_pass in 0..C {
529                let _client_pass = client_pass;
530                for slot_pass in 0..W {
531                    let _slot_pass = slot_pass;
532                    let mut next_client = client;
533                    let mut next_slot = slot + b3(1);
534                    if slot.resize::<4>() + b4(1) >= b4(W as u128) {
535                        next_slot = b3(0);
536                        next_client = client + b3(1);
537                        if client.resize::<4>() + b4(1) >= b4(C as u128) {
538                            next_client = b3(0);
539                        }
540                    }
541                    if !copied
542                        && s.entries[client][slot].phase == b2(3)
543                        && s.entries[client][slot].replay
544                    {
545                        s.reply = s.entries[client][slot].result;
546                        s.entries[client][slot].replay = false;
547                        s.reply_client = next_client;
548                        s.reply_slot = next_slot;
549                        copied = true;
550                    }
551                    client = next_client;
552                    slot = next_slot;
553                }
554            }
555        }
556    }
557    o.retired = s.retired;
558    (o, s)
559}
rhdl_ubft/rpc/execution.rs

rhdl_ubft/rpc/
execution.rs
