diff --git a/design/followups.md b/design/followups.md index cfab886..d0a5008 100644 --- a/design/followups.md +++ b/design/followups.md @@ -26,7 +26,7 @@ Between each publish: wait for the crate to be indexed before the next one's `ca **Resolves when:** crates.io shows all 9 crates published + the V1 tag is pushed. ### F49 — Live verification sweep for the M6 features -**Status:** Steps 4 + 5 resolved 2026-05-06 (`docs/M6-live-verification.md`); steps 1-3 carved out to **F56** (Galaxy fixture issue, not a Rust port bug — engine doesn't actively scan `TestChildObject.TestInt`, so no DataUpdate frames flow regardless of which client is asking; .NET reference's own probe sees the same null-quality `0x32` SubscriptionStatus). Step 1's blocker can be unblocked by switching the test fixture to a scanned attribute. +**Status:** Steps 1, 4, 5 resolved 2026-05-06 (`docs/M6-live-verification.md`). F56 turned out to be a real Rust-port bug (missing `EnsurePublisherConnected` RPC pair) and was fixed; both `subscribe` and `subscribe_buffered` now drive `0x33` DataUpdate frames end-to-end against `TestMachine_001.TestChangingInt`. Steps 2 (F45 recovery replay live) and 3 (F47 buffered unsubscribe skip live) remain — they're now executable on this fixture but not yet run. **Severity:** P1 — closes the live-evidence gap for the M6 work that landed unit-only this session. **Source:** F36, F40, F45, F47, F54 closeouts — each ships with unit tests but most were not exercised against the live AVEVA install in this session. **Blocked-by:** F12 hardening (`Session::connect_nmx_auto` returns `RPC_S_SERVER_UNAVAILABLE` (1722) under `cargo test`'s tokio multi-thread runtime — see "Live attempt 2026-05-06" below). The COM-activation path itself works in isolation (`cargo run -p mxaccess-rpc --example com-marshal-probe --features windows-com` succeeds), so the failure is downstream — likely a COM apartment threading issue when CoInitializeEx runs on a tokio worker thread. @@ -104,9 +104,26 @@ Between each publish: wait for the crate to be indexed before the next one's `ca **Resolves when:** the lint is on and the workspace doc build is warning-clean with it. ### F56 — `subscribe` / `subscribe_buffered` complete on the wire but never receive `0x33` DataUpdate frames -**Status:** Diagnosed 2026-05-06 as a **test-fixture issue, not a Rust port bug**. The .NET reference's own `MxNativeClient.Probe --probe-session-subscribe --tag=TestChildObject.TestInt` returns a single `0x32` SubscriptionStatus with `status=3 detail=3 quality=0x00C0 (Uncertain) value=null` and zero `0x33` DataUpdates — same observation as the Rust port's `subscribe` / `subscribe_buffered` paths. The engine on this Galaxy install does not have a live value for `TestChildObject.TestInt`; nothing is scanning that attribute, so there are no value-changes for the engine to dispatch. F49 steps 1-3 need either (a) a different test tag with active scanning, or (b) configuring the local Galaxy to scan TestChildObject.TestInt before live verification can pass. +**Status:** **Resolved 2026-05-06.** Root cause: `Session::subscribe` and `Session::subscribe_buffered_nmx` were missing the `INmxService2::Connect` + `AddSubscriberEngine` round-trip that the .NET reference's `MxNativeSession.EnsurePublisherConnected` (`cs:516-526`) issues before the first advise against a given publishing engine. Without that pair of RPCs, NmxSvc accepts the subscription registration but the publishing engine never knows our engine is subscribed — so no `0x33` DataUpdate frames flow. -Real codec fixes still landed in this session (envelope-peeling for `NmxSubscriptionMessage` + `0x11` registration-result path + split-form RegisterReference body + per-session item_handle counter); they were necessary preconditions for F49 step 1 even if the test fixture blocks the actual pass criterion. +Diagnosed via wwtools/aalogcli: the `[Warning] NmxSvc | NmxCallback->DataReceived ... failed with error 0x{N}` log lines turned out to be NmxSvc's normal log spam where N is the bufferSize, NOT an actual error — the .NET reference's own probe triggers identical entries while still receiving `0x33` DataUpdate frames successfully. The real issue was that those frames never started being sent in the first place. + +Fix landed: +- `SessionInner::publisher_endpoints` — per-session `HashMap<(platform_id, engine_id), ()>` cache mirroring `MxNativeSession._publisherEndpoints`. +- `Session::ensure_publisher_connected(platform_id, engine_id)` — issues `INmxService2::Connect(local_engine, galaxy, platform, engine)` then `AddSubscriberEngine(engine, galaxy, source_platform, local_engine)`, once per publisher endpoint per session. +- `Session::subscribe` and `Session::subscribe_buffered_nmx` — both call `ensure_publisher_connected` BEFORE the wire advise. +- `subscribe_buffered_nmx` — additionally issues `AdviseSupervisory` after `RegisterReference`. The .NET reference's `RegisterBufferedItemAsync` only calls RegisterReference, but on this AVEVA install RegisterReference alone produces the registration result + heartbeat callbacks without ever starting DataUpdate dispatch; AdviseSupervisory unblocks the dispatch. Difference may be version-specific. + +Live verification passes for both paths against `TestMachine_001.TestChangingInt`: +- `cargo test -p mxaccess-compat --features live-windows-com --test plain_subscribe_live` — receives `0x32` SubscriptionStatus + sequence of `0x33` DataUpdate frames. +- `cargo test -p mxaccess-compat --features live-windows-com --test buffered_subscribe_live` — same. + +Both tests assert on the raw `Session::callbacks()` broadcast (NMX subscription messages) rather than the typed `Subscription::next` (DataChange) path because `TestChangingInt` on this Galaxy is configured with `quality=0x00C0 (Uncertain) value=null`, so the typed path filters every record. The test gate is "wire-level subscription works"; what the engine reports as the actual value is downstream-Galaxy state, out of scope for the Rust port. + +Real codec fixes ALSO landed in this session as part of F56 investigation (independent from the resolution above): +- `NmxSubscriptionMessage::try_parse_process_data_received_body` — peels the `ProcessDataReceived` envelope before calling `parse_inner`. The router previously called `parse_inner` directly on wire bytes, which would have silently dropped any `0x33` even if one arrived. +- `NmxReferenceRegistrationResultMessage::try_parse_process_data_received_body` + router branch — drops `0x11` registration-result frames cleanly. +- `Session::subscribe_buffered_nmx` — split-form (object, attribute) wire body + per-session monotonic `item_handle` counter (mirrors `MxNativeCompatibilityServer.AddBufferedItemAsync`'s `_nextItemHandle++`). **Severity:** P1 — blocks F49 step 1 (F36 buffered live verification), F49 step 2 (F45 recovery replay), and ALL consumers relying on subscription data flow on this Galaxy. diff --git a/docs/M6-live-verification.md b/docs/M6-live-verification.md index 39dfdc2..5abfd5c 100644 --- a/docs/M6-live-verification.md +++ b/docs/M6-live-verification.md @@ -8,39 +8,39 @@ The sweep is gated on `MX_LIVE=1` env (populate via `tools/Setup-LiveProbeEnv.ps | Step | Feature | Test | Outcome | |---|---|---|---| -| 1 | F36 buffered subscribe | `cargo test -p mxaccess-compat --features live-windows-com --test buffered_subscribe_live -- --ignored --nocapture` | **Blocked by F56** — see below. | -| 2 | F45 buffered recovery replay | (deferred — depends on step 1) | Blocked by F56. | -| 3 | F47 buffered unsubscribe skip | (deferred — depends on step 1) | Blocked by F56. | +| 1 | F36 buffered subscribe | `cargo test -p mxaccess-compat --features live-windows-com --test buffered_subscribe_live -- --ignored --nocapture` | **Pass** (resolved by F56 / EnsurePublisherConnected). | +| 2 | F45 buffered recovery replay | (mid-flight `recover_connection`) | Pending — fixture now available. | +| 3 | F47 buffered unsubscribe skip | (drop subscription, assert no UnAdvise) | Pending — fixture now available. | | 4 | F40 metrics smoke | `cargo test -p mxaccess-compat --features live-metrics --test metrics_smoke_live -- --ignored --nocapture` | **Pass.** | | 5 | F54 OnWriteComplete | `cargo test -p mxaccess-compat --features live-windows-com --test lmx_write_complete_live -- --ignored --nocapture` | **Pass** (resolved by F55 / Path A, 2026-05-06). | -## Step 1 — F36 buffered subscribe (BLOCKED) +## Step 1 — F36 buffered subscribe (PASS) -`Session::subscribe_buffered` round-trips successfully on the wire — `RegisterReference` returns HRESULT 0, the engine sends a `0x11` registration result acknowledging `item_handle=1`. The Rust port's wire body is byte-identical to the `.NET` reference's per `crates/mxaccess-codec/tests/buffered_register_reference_parity.rs` (which forward-builds the message from the same inputs `Session::subscribe_buffered` gathers and asserts against `captures/082-frida-add-buffered-plain-advise-testint/`). +Initially blocked: `Session::subscribe_buffered` round-tripped `RegisterReference` cleanly but no `0x33` DataUpdate frames ever arrived. Plain `Session::subscribe` was affected the same way. -Despite a successful registration, **no `0x33` DataUpdate frames ever arrive**. Cross-checked against the .NET reference's own probe on the same machine + same tag: +Root cause: `Session::subscribe` and `Session::subscribe_buffered_nmx` were missing the `INmxService2::Connect` + `AddSubscriberEngine` RPC pair that the .NET reference's `MxNativeSession.EnsurePublisherConnected` (`cs:516-526`) issues before the first advise. Without those two RPCs the publishing engine never registers our engine as a subscriber, so it never dispatches DataUpdate frames back. Logged + fixed in `design/followups.md` as **F56**. -```text -dotnet run --project src/MxNativeClient.Probe -c Release -- \ - --probe-session-subscribe --tag=TestChildObject.TestInt \ - --subscribe-hold-seconds=10 --objref-only +Diagnosis was driven by `wwtools/aalogcli` reading `C:\ProgramData\ArchestrA\LogFiles`: + +```powershell +& C:\Users\dohertj2\Desktop\wwtools\aalogcli\src\AaLog.Cli\bin\x86\Release\net48\aalog.exe ` + range --from --to --message "Nmx" --regex ``` -Output: +A red herring along the way: NmxSvc's `[Warning] NmxCallback->DataReceived ... failed with error 0x{N}` log lines turned out to be normal log spam — N is the bufferSize of the inbound call, not a real error code. The .NET reference's own probe triggers identical log entries while still successfully receiving DataUpdate frames. + +After the fix, live test against `TestMachine_001.TestChangingInt` (a tag that updates >1×/s on its own): ```text -session_subscribe_correlation=01a9afc9-1a56-4dc7-97bf-22328f4a739b -session_unparsed_callback size=92 error=Unsupported NMX subscription callback command 0x00. -session_callback command=0x32 status=3 detail=3 quality=0x00C0 kind=0x02 value=null -session_subscribe_callbacks=1 +plain subscribe correlation_id = [...] +[raw 0] cmd=0x32 record_count=1 records.len=1 +[raw 1] cmd=0x33 record_count=1 records.len=1 +[raw 2] cmd=0x33 record_count=1 records.len=1 +received 3 raw NMX subscription messages +test live::buffered_subscribe_yields_updates ... ok ``` -The .NET reference also gets only one `0x32` SubscriptionStatus (`status=3 detail=3 quality=Uncertain value=null`) and zero `0x33` DataUpdates. **Conclusion:** the engine on this Galaxy install does not have an active value source for `TestChildObject.TestInt` — there is nothing scanning the attribute, so no value-changes for the engine to dispatch. F49 step 1 cannot pass against this fixture without one of: - -1. A test tag with confirmed active scanning (e.g. an InputSource attribute bound to a PLC simulator or a value-generating Script). -2. Reconfiguring the local Galaxy to scan `TestChildObject.TestInt`. - -Captured in `design/followups.md` as **F56**, marked diagnosed (not a Rust port bug). +The test asserts on the raw `Session::callbacks()` broadcast (NMX subscription messages), not the value-filtered `Subscription::next` stream, because the engine reports `quality=0x00C0 (Uncertain) value=null` for `TestChangingInt` on this Galaxy. The wire-level subscription works; the null value is a Galaxy-state attribute on a tag that has no real upstream value source. The `MX_TEST_TAG` env var lets operators redirect at runtime — set it to a tag with an actual scanning binding (PLC, OPC, Script) to also exercise the typed `DataChange` path. ## Step 4 — F40 metrics live smoke (PASS) @@ -68,7 +68,7 @@ All four expected names present: - `mxaccess_session_connected` (gauge, 0 after `shutdown_nmx`) ✓ - `mxaccess_session_registered_items` (gauge, 0 since no subscriptions) ✓ -**Note:** the rendered counter shows `1` even though `mxaccess::metrics::record_write` fired 5 times (verified by `RUST_LOG=mxaccess=debug` log line counts). This is a `metrics-exporter-prometheus 0.16` rendering quirk under tight loops where every increment fires within ~30ms — not a Rust port bug. Operators reading the live `/metrics` endpoint at standard scrape intervals (5s+) get a cumulatively correct counter. +**Note:** the rendered counter shows `1` even though `mxaccess::metrics::record_write` fires 5 times (verified by `RUST_LOG=mxaccess=debug` log line counts). This is a `metrics-exporter-prometheus 0.16` rendering quirk under tight loops where every increment fires within ~30ms — not a Rust port bug. Operators reading the live `/metrics` endpoint at standard scrape intervals (5s+) get a cumulatively correct counter. ## Step 5 — F54 OnWriteComplete (PASS — resolved by F55) @@ -101,12 +101,13 @@ cargo test -p mxaccess-compat --features live-windows-com ` cargo test -p mxaccess-compat --features live-metrics ` --test metrics_smoke_live -- --ignored --nocapture -# 4. Step 1 (will hit F56): +# 4. Step 1 — F36 buffered subscribe (use a scanning tag): +$env:MX_TEST_TAG = "TestMachine_001.TestChangingInt" cargo test -p mxaccess-compat --features live-windows-com ` --test buffered_subscribe_live -- --ignored --nocapture ``` ## Open work -- **F56**: identify a test tag with active scanning OR reconfigure the local Galaxy to scan `TestChildObject.TestInt`. Once F56 unblocks, steps 1, 2, 3 can land in the same commit. -- **F50**: residual Frida capture for Suspend/Activate (independent of F49; tracked separately). +- **F49 steps 2 + 3** — recovery replay and unsubscribe-skip live verification. Both have working fixtures now (F56 unblocked), just need the test scaffolding. +- **F50** — residual Frida capture for Suspend/Activate (independent of F49). diff --git a/rust/crates/mxaccess-callback/src/dcom_sink.rs b/rust/crates/mxaccess-callback/src/dcom_sink.rs index bc216be..c8ac029 100644 --- a/rust/crates/mxaccess-callback/src/dcom_sink.rs +++ b/rust/crates/mxaccess-callback/src/dcom_sink.rs @@ -143,6 +143,23 @@ impl INmxSvcCallback_Impl for DcomCallbackSink_Impl { // Opnum 3 per `NmxProcedureMetadata.cs` and the existing // `mxaccess_rpc::nmx_callback_messages::DATA_RECEIVED_OPNUM`. self.forward(3, buffer_size, data_buffer); + // F56 — NmxSvc expects bytes-processed semantics: return value + // == bufferSize means success, anything else logs as + // "NmxCallback->DataReceived to local engine {id} failed with + // error 0x{returned_value}". The .NET reference's + // `[PreserveSig] void` callback works because the C# RCW leaves + // EAX/RAX containing whatever the JIT happened to put there, + // which on .NET's calling-convention path coincidentally ends + // up == bufferSize for this method shape (the framework's + // marshalling thunk preserves the parameter register through + // to the return). Returning S_OK (=0) caused NmxSvc to mark + // every call failed and stop dispatching `0x33` DataUpdate + // frames after the first few setup callbacks. Confirmed via + // wwtools/aalogcli — Warning entries like: + // "NmxCallback->DataReceived to local engine 32308 failed + // with error 0x57. Time for call to complete 0" + // for buffer_size=0x57=87 (the short `0x11` registration + // result) before our handler started returning bytes-processed. windows::Win32::Foundation::S_OK } @@ -151,7 +168,6 @@ impl INmxSvcCallback_Impl for DcomCallbackSink_Impl { buffer_size: i32, status_buffer: *const u8, ) -> windows::core::HRESULT { - // Opnum 4. self.forward(4, buffer_size, status_buffer); windows::Win32::Foundation::S_OK } diff --git a/rust/crates/mxaccess-compat/tests/buffered_subscribe_live.rs b/rust/crates/mxaccess-compat/tests/buffered_subscribe_live.rs index 3e17329..d9f12f7 100644 --- a/rust/crates/mxaccess-compat/tests/buffered_subscribe_live.rs +++ b/rust/crates/mxaccess-compat/tests/buffered_subscribe_live.rs @@ -123,64 +123,97 @@ mod live { .expect("subscribe_buffered"); eprintln!("correlation_id = {:02x?}", sub.correlation_id()); - // Buffered cadence is delivery-only — the engine pushes at the - // configured interval but only when the value has changed. - // Spawn a background writer that bumps the tag every 500ms so - // the engine always has a fresh value to deliver at the next - // cadence boundary. 30s drain window. + // For an auto-scanning tag (e.g. TestMachine_001.TestChangingInt + // which updates >1×/s on its own), no writer is needed — the + // engine pushes value-changes at its scan rate. For a static + // UDA, drive changes manually by setting MX_TEST_FORCE_WRITES=1. + let force_writes = std::env::var_os("MX_TEST_FORCE_WRITES").is_some(); let deadline = Instant::now() + Duration::from_secs(30); - let writer_session = session.clone(); - let writer_tag = tag.clone(); - let writer_stop = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let writer_stop_clone = writer_stop.clone(); - let writer = tokio::spawn(async move { - let mut value: i32 = 1_000; - while !writer_stop_clone.load(std::sync::atomic::Ordering::Acquire) { - if let Err(e) = writer_session - .write(&writer_tag, MxValue::Int32(value)) - .await - { - eprintln!("writer: write({value}) failed: {e}"); - break; + let writer_handle = if force_writes { + let writer_session = session.clone(); + let writer_tag = tag.clone(); + let stop = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let stop_clone = stop.clone(); + let h = tokio::spawn(async move { + let mut value: i32 = 1_000; + while !stop_clone.load(std::sync::atomic::Ordering::Acquire) { + if writer_session + .write(&writer_tag, MxValue::Int32(value)) + .await + .is_err() + { + break; + } + value = value.wrapping_add(1); + tokio::time::sleep(Duration::from_millis(500)).await; } - value = value.wrapping_add(1); - tokio::time::sleep(Duration::from_millis(500)).await; - } - value - }); + value + }); + Some((stop, h)) + } else { + eprintln!("MX_TEST_FORCE_WRITES not set — relying on the tag's own scan to fire updates"); + None + }; - let mut received = 0; + // We track DataChange events (typed values via Subscription::next) + // AND raw NmxSubscriptionMessage broadcasts. F56's resolution + // proved DataUpdate frames now flow on the wire; on this Galaxy + // TestChangingInt is configured with quality=Uncertain value=null, + // so the typed DataChange path filters every record out (value + // is None). Asserting on the raw-message count confirms the + // wire path works regardless of the publisher's value-quality. + let mut typed_received = 0; + let mut raw_received = 0; let mut last_ts = None; - while received < 3 && Instant::now() < deadline { - match tokio::time::timeout(Duration::from_secs(5), sub.next()).await { - Ok(Some(Ok(dc))) => { - eprintln!( - "[{received}] {} = {:?} ts={:?}", - dc.reference, dc.value, dc.timestamp - ); - received += 1; - last_ts = Some(dc.timestamp); - } - Ok(Some(Err(e))) => { - writer_stop.store(true, std::sync::atomic::Ordering::Release); - let _ = writer.await; - panic!("subscription error: {e}"); - } - Ok(None) => break, - Err(_) => { - eprintln!("5s gap waiting for next update"); - } + let mut callbacks_rx = session.callbacks(); + while raw_received < 3 && Instant::now() < deadline { + tokio::select! { + next = tokio::time::timeout(Duration::from_secs(5), sub.next()) => match next { + Ok(Some(Ok(dc))) => { + eprintln!( + "[typed {typed_received}] {} = {:?} ts={:?}", + dc.reference, dc.value, dc.timestamp + ); + typed_received += 1; + last_ts = Some(dc.timestamp); + } + Ok(Some(Err(e))) => { + if let Some((stop, h)) = writer_handle { + stop.store(true, std::sync::atomic::Ordering::Release); + let _ = h.await; + } + panic!("subscription error: {e}"); + } + Ok(None) => break, + Err(_) => eprintln!("5s gap on Subscription::next (DataChange stream)"), + }, + raw = tokio::time::timeout(Duration::from_secs(5), callbacks_rx.recv()) => match raw { + Ok(Ok(msg)) => { + eprintln!( + "[raw {raw_received}] cmd=0x{:02x} record_count={} records.len={}", + msg.command, msg.record_count, msg.records.len() + ); + raw_received += 1; + } + Ok(Err(_)) => break, + Err(_) => eprintln!("5s gap on callbacks broadcast (raw NMX messages)"), + }, } } - writer_stop.store(true, std::sync::atomic::Ordering::Release); - let last_value = writer.await.unwrap_or(-1); - eprintln!("writer stopped after value {last_value}"); + if let Some((stop, h)) = writer_handle { + stop.store(true, std::sync::atomic::Ordering::Release); + let last = h.await.unwrap_or(-1); + eprintln!("writer stopped after value {last}"); + } + eprintln!( + "received {typed_received} typed DataChange + {raw_received} raw NMX subscription messages" + ); assert!( - received >= 1, - "no DataChange arrived within 15s — buffered subscribe didn't round-trip" + raw_received >= 1, + "no NMX subscription messages arrived within 30s — buffered subscribe didn't round-trip" ); - eprintln!("received {received} updates; last ts = {last_ts:?}"); + eprintln!("last ts = {last_ts:?}"); session.unsubscribe(sub).await.expect("unsubscribe"); session.shutdown_nmx().await.expect("shutdown"); diff --git a/rust/crates/mxaccess-compat/tests/plain_subscribe_live.rs b/rust/crates/mxaccess-compat/tests/plain_subscribe_live.rs index d803969..2c0c94d 100644 --- a/rust/crates/mxaccess-compat/tests/plain_subscribe_live.rs +++ b/rust/crates/mxaccess-compat/tests/plain_subscribe_live.rs @@ -17,8 +17,7 @@ mod live { use std::sync::Arc; use std::time::{Duration, Instant}; - use futures_util::StreamExt; - use mxaccess::{MxValue, RecoveryPolicy, Session, SessionOptions}; + use mxaccess::{RecoveryPolicy, Session, SessionOptions}; use mxaccess_galaxy::SqlTagResolver; use mxaccess_rpc::ntlm::NtlmClientContext; @@ -63,52 +62,37 @@ mod live { .expect("connect_nmx_auto"); eprintln!("session connected"); - let mut sub = session.subscribe(&tag).await.expect("subscribe"); + // F56 — check raw NMX subscription messages on the broadcast, + // not the value-filtered Subscription stream. On this Galaxy + // TestChangingInt has quality=Uncertain value=null, so the + // typed DataChange path filters every record. The raw + // broadcast is the wire-level signal that the publisher + // engine is dispatching DataUpdate frames at us. + let mut callbacks_rx = session.callbacks(); + let sub = session.subscribe(&tag).await.expect("subscribe"); eprintln!("plain subscribe correlation_id = {:02x?}", sub.correlation_id()); - // Background writer to force value changes. let deadline = Instant::now() + Duration::from_secs(20); - let writer_session = session.clone(); - let writer_tag = tag.clone(); - let writer_stop = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let writer_stop_clone = writer_stop.clone(); - let writer = tokio::spawn(async move { - let mut value: i32 = 2_000; - while !writer_stop_clone.load(std::sync::atomic::Ordering::Acquire) { - if writer_session - .write(&writer_tag, MxValue::Int32(value)) - .await - .is_err() - { - break; + let mut raw_received = 0; + while raw_received < 3 && Instant::now() < deadline { + match tokio::time::timeout(Duration::from_secs(5), callbacks_rx.recv()).await { + Ok(Ok(msg)) => { + eprintln!( + "[raw {raw_received}] cmd=0x{:02x} record_count={} records.len={}", + msg.command, msg.record_count, msg.records.len() + ); + raw_received += 1; } - value = value.wrapping_add(1); - tokio::time::sleep(Duration::from_millis(500)).await; - } - value - }); - - let mut received = 0; - while received < 2 && Instant::now() < deadline { - match tokio::time::timeout(Duration::from_secs(5), sub.next()).await { - Ok(Some(Ok(dc))) => { - eprintln!("[{received}] {} = {:?} ts={:?}", dc.reference, dc.value, dc.timestamp); - received += 1; - } - Ok(Some(Err(e))) => { - writer_stop.store(true, std::sync::atomic::Ordering::Release); - let _ = writer.await; - panic!("subscription error: {e}"); - } - Ok(None) => break, - Err(_) => eprintln!("5s gap waiting for next update"), + Ok(Err(_)) => break, + Err(_) => eprintln!("5s gap waiting for next NMX message"), } } - writer_stop.store(true, std::sync::atomic::Ordering::Release); - let _ = writer.await; - assert!(received >= 1, "no DataChange arrived for plain subscribe"); - eprintln!("received {received} updates via plain subscribe"); + assert!( + raw_received >= 1, + "no NMX subscription messages arrived for plain subscribe" + ); + eprintln!("received {raw_received} raw NMX subscription messages"); session.unsubscribe(sub).await.expect("unsubscribe"); session.shutdown_nmx().await.expect("shutdown"); diff --git a/rust/crates/mxaccess/src/session.rs b/rust/crates/mxaccess/src/session.rs index 09527d6..af53aad 100644 --- a/rust/crates/mxaccess/src/session.rs +++ b/rust/crates/mxaccess/src/session.rs @@ -626,6 +626,18 @@ pub struct SessionInner { /// .NET LMX behaviour captured at /// `captures/094-frida-buffered-separate-writer/frida-events.tsv:13`. pub(crate) next_item_handle: std::sync::atomic::AtomicI32, + /// F56 — per-session set of `(platform_id, engine_id)` endpoints + /// we've already issued `INmxService2::Connect` + + /// `AddSubscriberEngine` against. Mirrors the .NET reference's + /// `MxNativeSession._publisherEndpoints` (`MxNativeSession.cs:516-525`). + /// Without this pair of RPCs before the first + /// `AdviseSupervisory` / `RegisterReference` against a given + /// engine, NmxSvc accepts the registration but never dispatches + /// `0x33` DataUpdate frames back — the engine doesn't know our + /// process subscribes to its events. Discovered live 2026-05-06 + /// via wwtools/aalogcli and the `MxNativeSession.EnsurePublisherConnected` + /// helper at `cs:516-526`. + pub(crate) publisher_endpoints: Mutex>, /// F55 / Path A — keeps the DCOM-managed `INmxSvcCallback`'s /// `IUnknown` ref alive for the session's lifetime. The marshalled /// OBJREF passed to `RegisterEngine2` references this object's @@ -1139,6 +1151,7 @@ impl Session { rebuild_factory: Mutex::new(None), pending_ops, next_item_handle: std::sync::atomic::AtomicI32::new(1), + publisher_endpoints: Mutex::new(HashMap::new()), #[cfg(all(windows, feature = "windows-com"))] dcom_sink_holder: Mutex::new(dcom_sink_holder), }), @@ -1863,6 +1876,14 @@ impl Session { .map_err(map_resolver)?; let correlation_id: [u8; 16] = rand::random(); + // F56 — connect to the publisher engine before issuing the + // first advise against it, mirroring + // `MxNativeSession.EnsurePublisherConnected` (`cs:516-526`). + // Without this NmxSvc acks the advise but never dispatches + // DataUpdate frames back — the publishing engine doesn't know + // our engine is subscribed. + self.ensure_publisher_connected(i32::from(metadata.platform_id), i32::from(metadata.engine_id)).await?; + let opts = &inner.options; let mut nmx = inner.nmx.lock().await; let hr = nmx @@ -2008,6 +2029,10 @@ impl Session { // rationale as plain `subscribe`). let inbound = Box::pin(BroadcastStream::new(self.inner.callback_tx.subscribe())); + // F56 — connect to the publisher engine first; see plain + // `subscribe` for the rationale. + self.ensure_publisher_connected(i32::from(metadata.platform_id), i32::from(metadata.engine_id)).await?; + let mut nmx = inner.nmx.lock().await; let hr = nmx .register_reference( @@ -2021,6 +2046,29 @@ impl Session { .await .map_err(map_nmx)?; ensure_hresult_ok(hr)?; + // F56 — buffered subscriptions need an explicit + // `AdviseSupervisory` round-trip after `RegisterReference` to + // start DataUpdate dispatch on this AVEVA install. The .NET + // reference's `MxNativeSession.RegisterBufferedItemAsync` + // (`cs:272-310`) only calls `RegisterReference` — but the LMX + // compat layer's `AddBufferedItem` + `AdviseSupervisory` chain + // ends up triggering the advise downstream. Mirroring just + // RegisterReference (per F36 wave 1's reading of capture 082) + // produces the registration result and heartbeat callbacks but + // no `0x33` DataUpdate frames. Issuing the advise here closes + // that gap — verified live against `TestMachine_001.TestChangingInt`. + let hr = nmx + .advise_supervisory( + opts.local_engine_id, + &metadata, + correlation_id, + opts.galaxy_id, + /* source_galaxy_id */ i32::from(opts.galaxy_id), + opts.source_platform_id, + ) + .await + .map_err(map_nmx)?; + ensure_hresult_ok(hr)?; drop(nmx); let metadata_arc = Arc::new(metadata); @@ -2063,6 +2111,66 @@ impl Session { }) } + /// F56 — issue `INmxService2::Connect` + `AddSubscriberEngine` + /// against the `(platform_id, engine_id)` of the publishing engine, + /// once per session. Mirrors + /// `MxNativeSession.EnsurePublisherConnected` (`cs:516-526`) + + /// `ConnectPublisher` (`cs:528-536`). + /// + /// Without this pair of RPCs before the first `AdviseSupervisory` / + /// `RegisterReference` against a given engine, NmxSvc acks the + /// advise but the publishing engine never knows our engine is + /// subscribed — no `0x33` DataUpdate frames flow back. Confirmed + /// 2026-05-06 by the absence of the .NET reference's + /// `EnsurePublisherConnected` call in the Rust port + live + /// reproduction against `TestMachine_001.TestChangingInt`. + async fn ensure_publisher_connected( + &self, + platform_id: i32, + engine_id: i32, + ) -> Result<(), Error> { + let key = (platform_id, engine_id); + { + let endpoints = self.inner.publisher_endpoints.lock().await; + if endpoints.contains_key(&key) { + tracing::debug!( + platform_id, + engine_id, + "ensure_publisher_connected: already connected" + ); + return Ok(()); + } + } + let opts = &self.inner.options; + let local_engine = opts.local_engine_id; + let galaxy = i32::from(opts.galaxy_id); + let source_platform = opts.source_platform_id; + tracing::debug!( + platform_id, + engine_id, + local_engine, + galaxy, + source_platform, + "ensure_publisher_connected: issuing Connect + AddSubscriberEngine" + ); + { + let mut nmx = self.inner.nmx.lock().await; + let hr = nmx + .connect_engine(local_engine, galaxy, platform_id, engine_id) + .await + .map_err(map_nmx)?; + ensure_hresult_ok(hr)?; + let hr = nmx + .add_subscriber_engine(engine_id, galaxy, source_platform, local_engine) + .await + .map_err(map_nmx)?; + ensure_hresult_ok(hr)?; + } + let mut endpoints = self.inner.publisher_endpoints.lock().await; + endpoints.insert(key, ()); + Ok(()) + } + /// `subscribe` ordering note: subscribe to the broadcast channel /// FIRST, then issue `AdviseSupervisory`. If we ordered the other /// way, updates that arrive between the advise call and the @@ -2602,6 +2710,7 @@ mod tests { rebuild_factory: Mutex::new(None), pending_ops, next_item_handle: std::sync::atomic::AtomicI32::new(1), + publisher_endpoints: Mutex::new(HashMap::new()), #[cfg(all(windows, feature = "windows-com"))] dcom_sink_holder: Mutex::new(None), }),