phoxal 0.32.4

Phoxal - production-oriented autonomous robot framework: the runtime engine and model (the api contract tree lives in phoxal-api, the typed bus in phoxal-bus).
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110

//! A malformed `y2026_9::telemetry::Process` payload, pushed through a live bus
//! subscription, must be surfaced as a decode error (counted, not silently
//! accepted) - and a well-formed one on the same key must still decode.
//!
//! `phoxal-api`'s round-trip tests only prove encode+decode are self-consistent
//! (both could drift together and still pass); this exercises the real
//! subscriber decode path (`spawn_subscription` -> `decode_sample` ->
//! `health().decode_errors`) with a hand-injected bad payload, the guarantee the
//! CLI telemetry consumer relies on. `phoxal-bus` cannot host this test: it must
//! not depend on `phoxal-api` (that would be circular), so the y2026_9 body is
//! only nameable from the engine crate up.

use std::sync::atomic::Ordering;
use std::time::Duration;

use phoxal::raw::{
    Bus, BusConfig, BusMetadata, Codec, LogicalTime, MessagePack, OwnerCap, Publisher, Source,
    Subscriber, encoding_string,
};
use phoxal_api::y2026_9;
use serial_test::serial;
use zenoh::bytes::{Encoding, ZBytes};

#[serial]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn y2026_9_process_subscription_counts_a_malformed_payload_as_a_decode_error() {
    let bus = Bus::open(BusConfig::in_process("test/telemetry-decode", "robot"))
        .await
        .expect("bus should open");

    // Client-side (subscribe) topic, exactly as the CLI telemetry consumer builds it.
    let sub_topic = y2026_9::topic::new().telemetry().process();
    let subscriber = Subscriber::<y2026_9::telemetry::Process>::new(&bus, &sub_topic, 8)
        .await
        .expect("subscription should declare");

    // A well-formed sample on the same key must decode and be delivered - proves
    // the subscription is live, so the decode-error assertion below can't be a
    // false pass from a dead subscription.
    let owner_topic = y2026_9::topic::internal::new(OwnerCap::__mint())
        .telemetry()
        .process();
    let publisher = Publisher::<y2026_9::telemetry::Process>::new(bus.clone(), &owner_topic)
        .expect("owner publisher should attach");
    let good = y2026_9::telemetry::Process {
        cpu_pct: 4.25,
        rss_bytes: 33_554_432,
        window_ns: 3_000_000_000,
    };
    publisher
        .publish_at(LogicalTime::new(0, 1), good.clone())
        .await
        .expect("good sample should publish");

    let received = tokio::time::timeout(Duration::from_secs(5), subscriber.recv())
        .await
        .expect("a well-formed sample should arrive before the deadline")
        .expect("recv should not error");
    assert_eq!(received.body, good);
    assert_eq!(
        bus.health().decode_errors.load(Ordering::Relaxed),
        0,
        "a well-formed sample must not be counted as a decode error"
    );

    // Now inject a corrupt body payload on the SAME key. The envelope (codec +
    // metadata attachment) is valid so it passes the fast-reject; only the body
    // fails to decode - the exact drift a round-trip test cannot catch. 0xc1 is
    // MessagePack's never-used marker byte.
    let full_key = bus.full_key(sub_topic.key());
    let metadata = BusMetadata {
        codec: MessagePack::ID.as_u8(),
        produced_at_ns: 2,
        epoch: 0,
        source: Source {
            participant: "malformed-injector".to_string(),
            incarnation: 0,
            sequence: 0,
        },
    };
    bus.session()
        .put(full_key.as_str(), ZBytes::from(vec![0xc1u8, 0xc1, 0xc1]))
        .encoding(Encoding::from(encoding_string(MessagePack::ID)))
        .attachment(metadata.encode())
        .await
        .expect("raw malformed put should enqueue");

    // The decode error is surfaced asynchronously by the subscription task.
    // Bounded wait (never an open-ended loop): poll the counter, fail on the
    // deadline rather than hang.
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    loop {
        if bus.health().decode_errors.load(Ordering::Relaxed) >= 1 {
            break;
        }
        assert!(
            tokio::time::Instant::now() < deadline,
            "the malformed payload was not counted as a decode error within the deadline"
        );
        tokio::time::sleep(Duration::from_millis(10)).await;
    }

    // The malformed sample must NOT have been delivered as a body.
    assert!(
        subscriber.try_recv().is_none(),
        "a payload that fails to decode must never be delivered to the subscriber"
    );

    bus.close().await.expect("bus should close");
}