From f332c130ed45f4766f6c9b5eb66e2f22f3145fe6 Mon Sep 17 00:00:00 2001 From: Ivan FB Date: Thu, 16 Jul 2026 16:24:46 +0200 Subject: [PATCH] feat(channels): add channelExists to reliable channels API Callers need to know whether a channel is open before create/send, and had no way to ask without inferring it from a "channel already exists" error. Exposed at both layers so FFI consumers get the same answer. "Exists" means currently held by the manager, i.e. between a successful createReliableChannel and closeChannel. Persisted SDS state outlives a close, so a closed channel reports false -- this is a liveness check, not a "was this channel ever created" check. Cherry-picked from chore/channel-exists, which forked before the nim-ffi 0.2.0 migration. The FFI proc is re-expressed in the typed model, which also lets it carry a real bool instead of a "true"/"false" string; the hand-written header hunk is dropped since that header is generated now. Closes #4039 Co-Authored-By: Claude Opus 4.8 --- library/channels_api/channel_api.nim | 10 +++++ library/rust_bindings/src/api.rs | 19 +++++++++ library/rust_bindings/src/ffi.rs | 1 + library/rust_bindings/src/types.rs | 6 +++ .../api/reliable_channel_manager_api.nim | 1 + .../channels/api/channel_lifecycle.nim | 6 +++ tests/channels/test_all.nim | 1 + tests/channels/test_channel_lifecycle.nim | 39 +++++++++++++++++++ 8 files changed, 83 insertions(+) create mode 100644 tests/channels/test_channel_lifecycle.nim diff --git a/library/channels_api/channel_api.nim b/library/channels_api/channel_api.nim index 47b52c438..5c5da1ad2 100644 --- a/library/channels_api/channel_api.nim +++ b/library/channels_api/channel_api.nim @@ -47,6 +47,16 @@ proc logosdeliveryChannelCreate*( return ok(string(id)) +proc logosdeliveryChannelExists*( + lib: LogosDelivery, channelIdStr: string +): Future[Result[bool, string]] {.ffi.} = + ## Whether the channel is currently held by the manager, i.e. between a + ## successful create and close. A missing channel is `false`, not an error. + requireChannels(lib, "ChannelExists"): + return err(errMsg) + + return ok(lib.reliableChannelManager.channelExists(ChannelId(channelIdStr))) + proc logosdeliveryChannelSend*( lib: LogosDelivery, channelIdStr: string, req: ChannelSendRequest ): Future[Result[string, string]] {.ffi.} = diff --git a/library/rust_bindings/src/api.rs b/library/rust_bindings/src/api.rs index c8301667b..7fcffe9a2 100644 --- a/library/rust_bindings/src/api.rs +++ b/library/rust_bindings/src/api.rs @@ -1384,6 +1384,25 @@ impl LogosDeliveryCtx { decode_cbor::(&raw_bytes) } + pub fn channel_exists(&self, channel_id_str: String) -> Result { + let req = LogosdeliveryChannelExistsReq { channel_id_str }; + let req_bytes = encode_cbor(&req)?; + let raw_bytes = ffi_call_sync(self.timeout, |cb, ud| unsafe { + ffi::logosdelivery_channel_exists(self.ptr, cb, ud, req_bytes.as_ptr(), req_bytes.len()) + })?; + decode_cbor::(&raw_bytes) + } + + pub async fn channel_exists_async(&self, channel_id_str: String) -> Result { + let req = LogosdeliveryChannelExistsReq { channel_id_str }; + let req_bytes = encode_cbor(&req)?; + let ptr = self.ptr as usize; + let raw_bytes = ffi_call_async(self.timeout, move |cb, ud| unsafe { + ffi::logosdelivery_channel_exists(ptr as *mut c_void, cb, ud, req_bytes.as_ptr(), req_bytes.len()) + }).await?; + decode_cbor::(&raw_bytes) + } + pub fn channel_send(&self, channel_id_str: String, req: ChannelSendRequest) -> Result { let req = LogosdeliveryChannelSendReq { channel_id_str, req }; let req_bytes = encode_cbor(&req)?; diff --git a/library/rust_bindings/src/ffi.rs b/library/rust_bindings/src/ffi.rs index 68f024d76..a11ce65fd 100644 --- a/library/rust_bindings/src/ffi.rs +++ b/library/rust_bindings/src/ffi.rs @@ -57,6 +57,7 @@ extern "C" { pub fn waku_filter_unsubscribe(ctx: *mut c_void, callback: FFICallback, user_data: *mut c_void, req_cbor: *const u8, req_cbor_len: usize) -> c_int; pub fn waku_filter_unsubscribe_all(ctx: *mut c_void, callback: FFICallback, user_data: *mut c_void, req_cbor: *const u8, req_cbor_len: usize) -> c_int; pub fn logosdelivery_channel_create(ctx: *mut c_void, callback: FFICallback, user_data: *mut c_void, req_cbor: *const u8, req_cbor_len: usize) -> c_int; + pub fn logosdelivery_channel_exists(ctx: *mut c_void, callback: FFICallback, user_data: *mut c_void, req_cbor: *const u8, req_cbor_len: usize) -> c_int; pub fn logosdelivery_channel_send(ctx: *mut c_void, callback: FFICallback, user_data: *mut c_void, req_cbor: *const u8, req_cbor_len: usize) -> c_int; pub fn logosdelivery_channel_close(ctx: *mut c_void, callback: FFICallback, user_data: *mut c_void, req_cbor: *const u8, req_cbor_len: usize) -> c_int; pub fn logosdelivery_add_event_listener(ctx: *mut c_void, event_name: *const c_char, callback: FFICallback, user_data: *mut c_void) -> u64; diff --git a/library/rust_bindings/src/types.rs b/library/rust_bindings/src/types.rs index 6f5e0d846..a33d0dd14 100644 --- a/library/rust_bindings/src/types.rs +++ b/library/rust_bindings/src/types.rs @@ -384,6 +384,12 @@ pub struct LogosdeliveryChannelCreateReq { pub encryption_str: String, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct LogosdeliveryChannelExistsReq { + #[serde(rename = "channelIdStr")] + pub channel_id_str: String, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct LogosdeliveryChannelSendReq { #[serde(rename = "channelIdStr")] diff --git a/logos_delivery/api/reliable_channel_manager_api.nim b/logos_delivery/api/reliable_channel_manager_api.nim index 8d3f63ff3..a43670665 100644 --- a/logos_delivery/api/reliable_channel_manager_api.nim +++ b/logos_delivery/api/reliable_channel_manager_api.nim @@ -10,6 +10,7 @@ type ReliableChannelApi* = concept c createReliableChannel( c, channelId = ChannelId, contentTopic = ContentTopic, senderId = SdsParticipantID ) is Result[ChannelId, string] + channelExists(c, channelId = ChannelId) is bool closeChannel(c, channelId = ChannelId) is Future[Result[void, string]] send(c, channelId = ChannelId, appPayload = seq[byte]) is Future[Result[RequestId, string]] diff --git a/logos_delivery/channels/api/channel_lifecycle.nim b/logos_delivery/channels/api/channel_lifecycle.nim index d110ceb72..bacce83d8 100644 --- a/logos_delivery/channels/api/channel_lifecycle.nim +++ b/logos_delivery/channels/api/channel_lifecycle.nim @@ -74,6 +74,12 @@ proc createReliableChannel*( self.channels[channelId] = chn return ok(channelId) +proc channelExists*(self: ReliableChannelManager, channelId: ChannelId): bool = + ## True while the channel is held by the manager, i.e. between a successful + ## `createReliableChannel` and `closeChannel`. Persisted SDS state for a + ## closed channel does not count as existing. + return self.channels.hasKey(channelId) + proc closeChannel*( self: ReliableChannelManager, channelId: ChannelId ): Future[Result[void, string]] {.async: (raises: []).} = diff --git a/tests/channels/test_all.nim b/tests/channels/test_all.nim index 04b448707..0c0855278 100644 --- a/tests/channels/test_all.nim +++ b/tests/channels/test_all.nim @@ -1,3 +1,4 @@ {.used.} +import ./test_channel_lifecycle import ./test_reliable_channel_send_receive diff --git a/tests/channels/test_channel_lifecycle.nim b/tests/channels/test_channel_lifecycle.nim new file mode 100644 index 000000000..c5cf810cf --- /dev/null +++ b/tests/channels/test_channel_lifecycle.nim @@ -0,0 +1,39 @@ +{.used.} + +import chronos, testutils/unittests +import brokers/broker_context + +import ../testlib/[common, testasync] + +import logos_delivery/api/conf/channels_conf +import logos_delivery/channels/reliable_channel_manager +import logos_delivery/channels/api/channel_lifecycle +import logos_delivery/channels/encryption/noop_encryption + +suite "Reliable Channel - lifecycle": + asyncTest "channelExists tracks create and close": + const + channelId = ChannelId("lifecycle-channel") + contentTopic = ContentTopic("/reliable-channel/test/proto") + + var manager: ReliableChannelManager + lockNewGlobalBrokerContext: + manager = ReliableChannelManager.new(ReliableChannelManagerConf()).expect( + "ReliableChannelManager.new" + ) + setNoopEncryption() + + check not manager.channelExists(channelId) + + discard manager + .createReliableChannel(channelId, contentTopic, SdsParticipantID("local")) + .expect("createReliableChannel") + + check manager.channelExists(channelId) + + ## An unrelated id must not be reported as existing. + check not manager.channelExists(ChannelId("other-channel")) + + (await manager.closeChannel(channelId)).expect("closeChannel") + + check not manager.channelExists(channelId)