mirror of
https://github.com/logos-messaging/logos-delivery-rust-bindings.git
synced 2026-07-30 06:53:29 +00:00
fix: rebuild when the vendor's Nim sources change
build.rs only invalidated on the generated rust_bindings, so editing any Nim source left the linked archive stale and the tests silently ran against the previous build -- which cost an hour of debugging a channel failure against instrumentation that was never in the binary. Also corrects channel_message_reaches_peer's ignore reason. The message is not lost in SDS: Persistency is a process-wide singleton that refuses re-targeting, so two in-process nodes share one SDS job whose rows are keyed by channel id alone. The receiver loads the sender's history and SDS rightly calls the message a replay. The channels API is fine; the test needs two processes. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
parent
aade2ed4d8
commit
dfedf2ba15
@ -116,21 +116,34 @@ async fn channel_create_send_close() {
|
|||||||
/// Two nodes sharing a channel: a channel message should cross the wire and
|
/// Two nodes sharing a channel: a channel message should cross the wire and
|
||||||
/// surface through the typed onChannelMessageReceived listener.
|
/// surface through the typed onChannelMessageReceived listener.
|
||||||
///
|
///
|
||||||
/// Ignored: it does not, and the cause is below this binding layer. The message
|
/// Ignored: this cannot work in-process, and the channels API is not at fault.
|
||||||
/// reaches the receiver's messaging layer with the right content topic and the
|
/// Persistency is a process-wide singleton keyed on its root dir, and it refuses
|
||||||
/// `RELIABLE-CHANNEL-API/1` marker -- both of the channel ingress filters pass --
|
/// to be re-targeted, so both nodes share one SDS persistence job whose rows are
|
||||||
/// and is then consumed silently by SDS `handleIncoming` (empty result means
|
/// keyed by channel id alone. The receiver's ensureChannel therefore loads the
|
||||||
/// parked or duplicate), so no event and no error is ever raised. logos-delivery
|
/// *sender's* history before the duplicate check, and SDS correctly rejects the
|
||||||
/// has no two-node channel test of its own; its suite fabricates the wire by
|
/// message as a replay -- silently, since a duplicate is not an error.
|
||||||
/// emitting a MessageReceivedEvent locally with a stand-in peer's SDS envelope,
|
|
||||||
/// so this path looks unexercised rather than regressed.
|
|
||||||
///
|
///
|
||||||
/// Un-ignore once the SDS ingress question is settled upstream: everything on
|
/// Everything either side of that works: autosharding, per-channel encryption,
|
||||||
/// this side (autosharding, per-channel encryption, send, connectivity) works.
|
/// the send, connectivity, the channel's ingress listener, and the typed event
|
||||||
|
/// path (proven by default_echo over the same route). Verified by instrumenting
|
||||||
|
/// the vendor: the listener fires, both filters pass, and SDS reports
|
||||||
|
/// `duplicate`.
|
||||||
|
///
|
||||||
|
/// Un-ignore by driving the two nodes as separate processes, or once SDS
|
||||||
|
/// persistence is keyed by participant as well as channel.
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
#[serial]
|
#[serial]
|
||||||
#[ignore = "channel ingress is dropped inside SDS; see the doc comment"]
|
#[ignore = "needs two processes: in-process nodes share SDS persistence; see the doc comment"]
|
||||||
async fn channel_message_reaches_peer() -> Result<(), String> {
|
async fn channel_message_reaches_peer() -> Result<(), String> {
|
||||||
|
// SDS state is persisted per channel id, so a fresh id keeps a previous
|
||||||
|
// run's causal history from parking this run's message.
|
||||||
|
let channel_id = format!(
|
||||||
|
"test-channel-{}",
|
||||||
|
std::time::SystemTime::now()
|
||||||
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
|
.unwrap()
|
||||||
|
.as_nanos()
|
||||||
|
);
|
||||||
let sender = new_node(60080);
|
let sender = new_node(60080);
|
||||||
let receiver = new_node(60090);
|
let receiver = new_node(60090);
|
||||||
|
|
||||||
@ -142,10 +155,14 @@ async fn channel_message_reaches_peer() -> Result<(), String> {
|
|||||||
|
|
||||||
// The raw message is observed too: it proves delivery reaches the messaging
|
// The raw message is observed too: it proves delivery reaches the messaging
|
||||||
// layer, which is what localises the failure to channel ingress.
|
// layer, which is what localises the failure to channel ingress.
|
||||||
let raw: Arc<Mutex<usize>> = Arc::new(Mutex::new(0));
|
let raw: Arc<Mutex<Vec<usize>>> = Arc::new(Mutex::new(Vec::new()));
|
||||||
let raw_cloned = raw.clone();
|
let raw_cloned = raw.clone();
|
||||||
receiver.add_on_message_received_listener(move |_| {
|
receiver.add_on_message_received_listener(move |e| {
|
||||||
*raw_cloned.lock().unwrap() += 1;
|
let len = base64::engine::general_purpose::STANDARD
|
||||||
|
.decode(&e.message.payload)
|
||||||
|
.map(|p| p.len())
|
||||||
|
.unwrap_or(0);
|
||||||
|
raw_cloned.lock().unwrap().push(len);
|
||||||
});
|
});
|
||||||
|
|
||||||
// A failed send reports itself here; without this the test could only time
|
// A failed send reports itself here; without this the test could only time
|
||||||
@ -169,7 +186,7 @@ async fn channel_message_reaches_peer() -> Result<(), String> {
|
|||||||
// participant ids.
|
// participant ids.
|
||||||
sender
|
sender
|
||||||
.channel_create_async(
|
.channel_create_async(
|
||||||
TEST_CHANNEL_ID.to_string(),
|
channel_id.clone(),
|
||||||
TEST_CONTENT_TOPIC.to_string(),
|
TEST_CONTENT_TOPIC.to_string(),
|
||||||
TEST_SENDER_ID.to_string(),
|
TEST_SENDER_ID.to_string(),
|
||||||
NOOP_ENCRYPTION.to_string(),
|
NOOP_ENCRYPTION.to_string(),
|
||||||
@ -177,7 +194,7 @@ async fn channel_message_reaches_peer() -> Result<(), String> {
|
|||||||
.await?;
|
.await?;
|
||||||
receiver
|
receiver
|
||||||
.channel_create_async(
|
.channel_create_async(
|
||||||
TEST_CHANNEL_ID.to_string(),
|
channel_id.clone(),
|
||||||
TEST_CONTENT_TOPIC.to_string(),
|
TEST_CONTENT_TOPIC.to_string(),
|
||||||
OTHER_SENDER_ID.to_string(),
|
OTHER_SENDER_ID.to_string(),
|
||||||
NOOP_ENCRYPTION.to_string(),
|
NOOP_ENCRYPTION.to_string(),
|
||||||
@ -197,7 +214,7 @@ async fn channel_message_reaches_peer() -> Result<(), String> {
|
|||||||
tokio::time::sleep(Duration::from_secs(5)).await;
|
tokio::time::sleep(Duration::from_secs(5)).await;
|
||||||
|
|
||||||
sender
|
sender
|
||||||
.channel_send_async(TEST_CHANNEL_ID.to_string(), channel_message(CHANNEL_PAYLOAD))
|
.channel_send_async(channel_id.clone(), channel_message(CHANNEL_PAYLOAD))
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
let mut delivered = None;
|
let mut delivered = None;
|
||||||
@ -214,21 +231,19 @@ async fn channel_message_reaches_peer() -> Result<(), String> {
|
|||||||
let msg_errors = msg_errors.lock().unwrap();
|
let msg_errors = msg_errors.lock().unwrap();
|
||||||
let raw = raw.lock().unwrap();
|
let raw = raw.lock().unwrap();
|
||||||
format!(
|
format!(
|
||||||
"never reached the peer; raw messages seen by receiver: {raw}; \
|
"never reached the peer; raw payload sizes seen by receiver: {raw:?}; \
|
||||||
channel errors: {errors:?}; messaging errors: {msg_errors:?}"
|
channel errors: {errors:?}; messaging errors: {msg_errors:?}"
|
||||||
)
|
)
|
||||||
})?;
|
})?;
|
||||||
assert_eq!(event.channel_id, TEST_CHANNEL_ID);
|
assert_eq!(event.channel_id, channel_id);
|
||||||
assert_eq!(event.sender_id, TEST_SENDER_ID);
|
assert_eq!(event.sender_id, TEST_SENDER_ID);
|
||||||
let payload = base64::engine::general_purpose::STANDARD
|
let payload = base64::engine::general_purpose::STANDARD
|
||||||
.decode(&event.payload)
|
.decode(&event.payload)
|
||||||
.map_err(|e| e.to_string())?;
|
.map_err(|e| e.to_string())?;
|
||||||
assert_eq!(payload, CHANNEL_PAYLOAD);
|
assert_eq!(payload, CHANNEL_PAYLOAD);
|
||||||
|
|
||||||
sender.channel_close_async(TEST_CHANNEL_ID.to_string()).await?;
|
sender.channel_close_async(channel_id.clone()).await?;
|
||||||
receiver
|
receiver.channel_close_async(channel_id).await?;
|
||||||
.channel_close_async(TEST_CHANNEL_ID.to_string())
|
|
||||||
.await?;
|
|
||||||
|
|
||||||
sender.stop_node_async().await?;
|
sender.stop_node_async().await?;
|
||||||
receiver.stop_node_async().await?;
|
receiver.stop_node_async().await?;
|
||||||
|
|||||||
@ -121,6 +121,17 @@ fn emit_link_flags(project_dir: &Path) {
|
|||||||
"cargo:rerun-if-changed={}",
|
"cargo:rerun-if-changed={}",
|
||||||
nwaku_path.join("library/rust_bindings/src").display()
|
nwaku_path.join("library/rust_bindings/src").display()
|
||||||
);
|
);
|
||||||
|
// The Nim sources the library is built from have to invalidate it too, or
|
||||||
|
// editing them leaves the linked archive stale and the tests silently run
|
||||||
|
// against the previous build.
|
||||||
|
println!(
|
||||||
|
"cargo:rerun-if-changed={}",
|
||||||
|
nwaku_path.join("library").display()
|
||||||
|
);
|
||||||
|
println!(
|
||||||
|
"cargo:rerun-if-changed={}",
|
||||||
|
nwaku_path.join("logos_delivery").display()
|
||||||
|
);
|
||||||
println!(
|
println!(
|
||||||
"cargo:rustc-link-search={}",
|
"cargo:rustc-link-search={}",
|
||||||
nwaku_path.join("build").display()
|
nwaku_path.join("build").display()
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user