Skip to main content

net/ffi/
aggregator.rs

1//! C FFI bindings for the aggregator-registry RPC client +
2//! fold-query client + channel-visibility setter
3//! (`SDK_AGGREGATOR_SUBNET_PLAN.md` stages 5 + 4-fold-query).
4//!
5//! Boundary conventions mirror `ffi::mesh`: opaque handles
6//! freed via dedicated `_free`, scalar ids as `u64`, JSON
7//! strings via `CString::into_raw` freed by the caller via
8//! `net_free_string`. Caller safety contract is identical to
9//! `ffi::mesh` / `ffi::cortex`; `clippy::missing_safety_doc`
10//! suppressed at the module level for the same rationale.
11#![allow(clippy::missing_safety_doc)]
12#![expect(
13    clippy::undocumented_unsafe_blocks,
14    reason = "module-wide FFI safety contract documented in ffi::mod.rs preamble"
15)]
16
17use std::ffi::{c_char, c_int, CStr, CString};
18use std::mem::ManuallyDrop;
19use std::time::Duration;
20
21use parking_lot::{Mutex as ParkingMutex, RwLock as ParkingRwLock};
22
23use super::handle_guard::{BeginFree, HandleGuard, FFI_HANDLE_FREE_DEADLINE};
24
25use crate::adapter::net::behavior::aggregator::{
26    FoldQueryClient, FoldQueryClientError, FoldQueryError, RegistryClient, RegistryClientError,
27    RegistryGroupSummary, RegistryRpcError, SummaryAnnouncement, DEFAULT_QUERY_DEADLINE,
28    DEFAULT_REGISTRY_DEADLINE,
29};
30use crate::adapter::net::{ChannelConfig, ChannelId, ChannelName, Visibility};
31
32use super::mesh::MeshNodeHandle;
33
34// ─── Error-kind discriminants (locked across SDKs) ───
35
36/// Server handler rejected: no summarizer registered under the
37/// requested fold kind. Only emitted by
38/// `net_fold_query_client_*` ops.
39pub const NET_REGISTRY_ERR_UNKNOWN_KIND: i32 = 7;
40
41/// `net_registry_client_*` op succeeded.
42pub const NET_REGISTRY_OK: i32 = 0;
43/// Transport-level failure (no route, timeout, server returned
44/// a non-Ok status before invoking the handler).
45pub const NET_REGISTRY_ERR_TRANSPORT: i32 = 1;
46/// Request serialization or response deserialization failed.
47pub const NET_REGISTRY_ERR_CODEC: i32 = 2;
48/// Server handler rejected: no template by that name.
49pub const NET_REGISTRY_ERR_UNKNOWN_TEMPLATE: i32 = 3;
50/// Server handler rejected: a group by that name is already
51/// registered.
52pub const NET_REGISTRY_ERR_DUPLICATE_GROUP_NAME: i32 = 4;
53/// Server handler rejected for a daemon-defined reason
54/// (config validation, replica spawn failed, etc.).
55pub const NET_REGISTRY_ERR_SPAWN_REJECTED: i32 = 5;
56/// Server doesn't accept dynamic spawn (read-only daemon).
57pub const NET_REGISTRY_ERR_SPAWN_NOT_SUPPORTED: i32 = 6;
58/// Server handler rejected `Scale`: no group by that name is
59/// registered on the target.
60pub const NET_REGISTRY_ERR_UNKNOWN_GROUP: i32 = 8;
61/// Server handler rejected `Scale` for a daemon-defined reason
62/// (template mismatch, replica spawn/stop failure, etc.).
63pub const NET_REGISTRY_ERR_SCALE_REJECTED: i32 = 9;
64/// Server doesn't accept dynamic scale (no scale handler
65/// installed).
66pub const NET_REGISTRY_ERR_SCALE_NOT_SUPPORTED: i32 = 10;
67/// Server refused: the caller is not an operator of the target
68/// daemon's aggregator registry. See `RegistryAdminPolicy`.
69pub const NET_REGISTRY_ERR_UNAUTHORIZED: i32 = 11;
70/// Caller-side error: a string argument wasn't valid UTF-8 or
71/// a pointer was null where one was required.
72pub const NET_REGISTRY_ERR_INVALID_ARGS: i32 = 99;
73
74// ─── Visibility discriminants ───
75
76/// Wire-equivalent of [`Visibility`]. Values are
77/// representation-stable across SDK releases — operator code
78/// referring to them by literal value (not just by name) stays
79/// correct. Mirrors every substrate variant 1-to-1; mirror order
80/// is sorted by tier-broadness for operator readability.
81#[repr(i32)]
82#[derive(Copy, Clone)]
83pub enum NetVisibility {
84    /// Mirrors [`Visibility::Global`] — visible everywhere.
85    Global = 0,
86    /// Mirrors [`Visibility::ParentVisible`].
87    ParentVisible = 1,
88    /// Mirrors [`Visibility::Exported`] — explicit per-subnet export list.
89    Exported = 2,
90    /// Mirrors [`Visibility::SubnetLocal`] — packets never leave the subnet.
91    SubnetLocal = 3,
92}
93
94impl NetVisibility {
95    fn from_raw(raw: i32) -> Option<Visibility> {
96        match raw {
97            0 => Some(Visibility::Global),
98            1 => Some(Visibility::ParentVisible),
99            2 => Some(Visibility::Exported),
100            3 => Some(Visibility::SubnetLocal),
101            _ => None,
102        }
103    }
104
105    /// Compile-time exhaustiveness check in the *opposite*
106    /// direction — every substrate [`Visibility`] variant must
107    /// have a wire-stable C ABI counterpart. If the substrate
108    /// gains a variant, this `match` stops compiling, forcing
109    /// the FFI maintainer to either add the discriminant + bump
110    /// the wire contract or explicitly accept the omission with
111    /// `_ => None`. Without this, [`from_raw`] would silently
112    /// reject the new variant and operator code referring to it
113    /// by literal value would see a NULL handle / ERR_INVALID
114    /// instead of a typed wire error.
115    #[allow(dead_code)] // existence is the check
116    fn to_raw(v: Visibility) -> NetVisibility {
117        match v {
118            Visibility::Global => NetVisibility::Global,
119            Visibility::ParentVisible => NetVisibility::ParentVisible,
120            Visibility::Exported => NetVisibility::Exported,
121            Visibility::SubnetLocal => NetVisibility::SubnetLocal,
122        }
123    }
124}
125
126// ─── Handle ───
127
128/// FFI handle for a [`RegistryClient`].
129///
130/// The inner client is wrapped in a `RwLock` so concurrent ops
131/// (entry points are called from many threads in async runtimes)
132/// can share read access while a `set_deadline` writer
133/// serializes. `last_error_detail` lives behind a separate
134/// `parking_lot::Mutex`; [`net_registry_last_error_detail`]
135/// returns an owned copy of its contents (freed with
136/// `net_free_string`), so the returned pointer never aliases this
137/// mutex-owned slot and can't dangle on overwrite/free.
138///
139/// Uses the same `HandleGuard` quiescing recipe as the
140/// cortex/mesh/redis-dedup handles (see [`super::handle_guard`]):
141/// the inner fields live in `ManuallyDrop`, every op gates on
142/// `guard.try_enter()`, and `_free` drains in-flight ops via
143/// `begin_free()` before dropping the inner — the box itself is
144/// leaked, never `Box::from_raw`'d, so a `_free` racing a
145/// concurrent op can't deallocate the lock out from under it.
146pub struct RegistryClientHandle {
147    client: ManuallyDrop<ParkingRwLock<RegistryClient>>,
148    last_error_detail: ManuallyDrop<ParkingMutex<Option<CString>>>,
149    guard: HandleGuard,
150}
151
152// ─── Constructor / free / builder ───
153
154/// Construct a `RegistryClient` against an existing
155/// [`MeshNodeHandle`]. Returns a handle the caller frees via
156/// [`net_registry_client_free`]. Returns NULL on null input.
157#[unsafe(no_mangle)]
158pub unsafe extern "C" fn net_registry_client_new(
159    mesh_handle: *mut MeshNodeHandle,
160) -> *mut RegistryClientHandle {
161    if mesh_handle.is_null() {
162        return std::ptr::null_mut();
163    }
164    // Gated clone of the mesh node — `None` means the mesh handle is
165    // being freed concurrently; surface a null handle rather than
166    // racing the inner out of `ManuallyDrop`.
167    let Some(mesh_arc) = (unsafe { super::mesh::mesh_node_arc(&*mesh_handle) }) else {
168        return std::ptr::null_mut();
169    };
170    let boxed = Box::new(RegistryClientHandle {
171        client: ManuallyDrop::new(ParkingRwLock::new(RegistryClient::new(mesh_arc))),
172        last_error_detail: ManuallyDrop::new(ParkingMutex::new(None)),
173        guard: HandleGuard::new(),
174    });
175    Box::into_raw(boxed)
176}
177
178/// Free a `RegistryClient` handle produced by
179/// [`net_registry_client_new`]. Idempotent on NULL and on a
180/// second call (the `begin_free` single-winner contract gates the
181/// inner drop). Quiesces in-flight ops before dropping the inner;
182/// the box stays leaked so a concurrent op can't UAF the handle.
183#[unsafe(no_mangle)]
184pub unsafe extern "C" fn net_registry_client_free(handle: *mut RegistryClientHandle) {
185    if handle.is_null() {
186        return;
187    }
188    let h: &RegistryClientHandle = unsafe { &*handle };
189    match h.guard.begin_free_detailed(FFI_HANDLE_FREE_DEADLINE) {
190        BeginFree::Drained => {
191            // SAFETY: drained; sole writable reference. Box leaked.
192            unsafe {
193                ManuallyDrop::drop(&mut (*handle).client);
194                ManuallyDrop::drop(&mut (*handle).last_error_detail);
195            }
196        }
197        // Benign repeat free — a prior call owns the inner; nothing
198        // to do and nothing leaked by this call.
199        BeginFree::AlreadyFreeing => {}
200        BeginFree::TimedOut => {
201            tracing::warn!(
202                "net_registry_client_free: in-flight ops did not drain within deadline; \
203                 leaking inner to avoid use-after-free"
204            );
205        }
206    }
207}
208
209/// Override the per-call deadline in milliseconds. `millis == 0`
210/// resets to the substrate default. Safe to call concurrently
211/// with in-flight ops; the writer takes the inner lock briefly
212/// and any concurrent reader either observes the old or the new
213/// deadline (no torn read).
214#[unsafe(no_mangle)]
215pub unsafe extern "C" fn net_registry_client_set_deadline(
216    handle: *mut RegistryClientHandle,
217    millis: u64,
218) {
219    if handle.is_null() {
220        return;
221    }
222    let h: &RegistryClientHandle = unsafe { &*handle };
223    let _op = match h.guard.try_enter() {
224        Some(op) => op,
225        None => return,
226    };
227    let deadline = if millis == 0 {
228        DEFAULT_REGISTRY_DEADLINE
229    } else {
230        Duration::from_millis(millis)
231    };
232    h.client.write().set_deadline_mut(deadline);
233}
234
235// ─── Op-handler internals ───
236//
237// Every public `net_registry_client_*` op shares the same six
238// steps: null-check, parse CStr args, snapshot the client under
239// the read lock, await the substrate call, classify+store-detail
240// on error, write the out param. The `dispatch_*` + `write_*`
241// helpers below capture each step once.
242
243/// Set `*out` if non-null and return the JSON pointer + status.
244/// Op handlers funnel every success / failure path through this
245/// so the null-check on `out_error_kind` is centralized.
246#[inline]
247unsafe fn write_kind(out: *mut c_int, kind: c_int) {
248    if !out.is_null() {
249        unsafe { *out = kind };
250    }
251}
252
253/// Read a NUL-terminated UTF-8 string argument and return an
254/// owned `String`, or set the out-param to `INVALID_ARGS` +
255/// return `None` if the pointer is null or the bytes aren't
256/// valid UTF-8.
257#[inline]
258unsafe fn cstr_arg(ptr: *const c_char, out: *mut c_int) -> Option<String> {
259    if ptr.is_null() {
260        unsafe { write_kind(out, NET_REGISTRY_ERR_INVALID_ARGS) };
261        return None;
262    }
263    match unsafe { CStr::from_ptr(ptr).to_str() } {
264        Ok(s) => Some(s.to_owned()),
265        Err(_) => {
266            unsafe { write_kind(out, NET_REGISTRY_ERR_INVALID_ARGS) };
267            None
268        }
269    }
270}
271
272/// Convert a JSON string into a heap-allocated `*mut c_char` the
273/// caller frees with `net_free_string`. Returns NULL + sets the
274/// out-param to `CODEC` if the string contains an embedded NUL.
275#[inline]
276unsafe fn json_to_raw(json: String, out: *mut c_int) -> *mut c_char {
277    match CString::new(json) {
278        Ok(s) => {
279            unsafe { write_kind(out, NET_REGISTRY_OK) };
280            s.into_raw()
281        }
282        Err(_) => {
283            unsafe { write_kind(out, NET_REGISTRY_ERR_CODEC) };
284            std::ptr::null_mut()
285        }
286    }
287}
288
289/// Funnel for any registry op that returns a JSON string.
290/// Takes a closure that produces `Result<String, RegistryClientError>`
291/// (the JSON-encoding step is the caller's responsibility because
292/// the substrate type varies per op).
293unsafe fn registry_op_json<F>(
294    handle: *mut RegistryClientHandle,
295    out_error_kind: *mut c_int,
296    op: F,
297) -> *mut c_char
298where
299    F: FnOnce(RegistryClient) -> Result<String, RegistryClientError>,
300{
301    if handle.is_null() {
302        unsafe { write_kind(out_error_kind, NET_REGISTRY_ERR_INVALID_ARGS) };
303        return std::ptr::null_mut();
304    }
305    let h: &RegistryClientHandle = unsafe { &*handle };
306    // Hold the guard ONLY long enough to clone the inner client (an
307    // Arc-backed handle that keeps the mesh node alive on its own).
308    // Bail with INVALID_ARGS (same shape as a NULL handle) if `_free`
309    // has begun. Dropping the guard before the blocking RPC means a
310    // concurrent `_free` never waits on the (caller-settable, possibly
311    // multi-second) op deadline — so it can't time out and leak the
312    // inner; the op simply completes against its own clone.
313    let client = match h.guard.try_enter() {
314        Some(_op) => h.client.read().clone(),
315        None => {
316            unsafe { write_kind(out_error_kind, NET_REGISTRY_ERR_INVALID_ARGS) };
317            return std::ptr::null_mut();
318        }
319    };
320    match op(client) {
321        Ok(json) => unsafe { json_to_raw(json, out_error_kind) },
322        Err(e) => {
323            let (kind, detail) = classify(&e);
324            // Re-enter only to record the detail; if the handle is now
325            // being freed, drop it (a freed handle won't be queried).
326            if let Some(_op) = h.guard.try_enter() {
327                store_error_detail(h, detail);
328            }
329            unsafe { write_kind(out_error_kind, kind) };
330            std::ptr::null_mut()
331        }
332    }
333}
334
335// ─── Operations ───
336
337/// Enumerate groups on `target_node_id`. Returns a JSON-encoded
338/// `[RegistryGroupSummaryJson]` string the caller frees via
339/// `net_free_string`. On error, writes the error kind to
340/// `*out_error_kind` and returns NULL.
341#[unsafe(no_mangle)]
342pub unsafe extern "C" fn net_registry_client_list(
343    handle: *mut RegistryClientHandle,
344    target_node_id: u64,
345    out_error_kind: *mut c_int,
346) -> *mut c_char {
347    if out_error_kind.is_null() {
348        return std::ptr::null_mut();
349    }
350    unsafe {
351        registry_op_json(handle, out_error_kind, |client| {
352            block_on(client.list(target_node_id)).map(|groups| groups_to_json(&groups))
353        })
354    }
355}
356
357/// Spawn a new group by referencing a daemon-side template.
358/// `template_name` + `group_name` are NUL-terminated UTF-8.
359#[unsafe(no_mangle)]
360pub unsafe extern "C" fn net_registry_client_spawn(
361    handle: *mut RegistryClientHandle,
362    target_node_id: u64,
363    template_name: *const c_char,
364    group_name: *const c_char,
365    replica_count: u8,
366    out_error_kind: *mut c_int,
367) -> *mut c_char {
368    let Some(template) = (unsafe { cstr_arg(template_name, out_error_kind) }) else {
369        return std::ptr::null_mut();
370    };
371    let Some(group) = (unsafe { cstr_arg(group_name, out_error_kind) }) else {
372        return std::ptr::null_mut();
373    };
374    unsafe {
375        registry_op_json(handle, out_error_kind, |client| {
376            block_on(client.spawn(target_node_id, template, group, replica_count))
377                .map(|summary| group_to_json(&summary))
378        })
379    }
380}
381
382/// Tear down a registered group by name. Returns `1` when the
383/// group existed and was stopped, `0` when no such group was
384/// registered, `-1` on transport / codec / invalid-args
385/// failure (consult `out_error_kind`).
386#[unsafe(no_mangle)]
387pub unsafe extern "C" fn net_registry_client_unregister(
388    handle: *mut RegistryClientHandle,
389    target_node_id: u64,
390    group_name: *const c_char,
391    out_error_kind: *mut c_int,
392) -> c_int {
393    if handle.is_null() {
394        unsafe { write_kind(out_error_kind, NET_REGISTRY_ERR_INVALID_ARGS) };
395        return -1;
396    }
397    let Some(group) = (unsafe { cstr_arg(group_name, out_error_kind) }) else {
398        return -1;
399    };
400    let h: &RegistryClientHandle = unsafe { &*handle };
401    // Guard held only for the clone — see `registry_op_json` for why
402    // the blocking RPC runs unguarded.
403    let client = match h.guard.try_enter() {
404        Some(_op) => h.client.read().clone(),
405        None => {
406            unsafe { write_kind(out_error_kind, NET_REGISTRY_ERR_INVALID_ARGS) };
407            return -1;
408        }
409    };
410    match block_on(client.unregister(target_node_id, group)) {
411        Ok(existed) => {
412            unsafe { write_kind(out_error_kind, NET_REGISTRY_OK) };
413            if existed {
414                1
415            } else {
416                0
417            }
418        }
419        Err(e) => {
420            let (kind, detail) = classify(&e);
421            if let Some(_op) = h.guard.try_enter() {
422                store_error_detail(h, detail);
423            }
424            unsafe { write_kind(out_error_kind, kind) };
425            -1
426        }
427    }
428}
429
430/// Get the operator-facing detail string for the most recent
431/// non-OK op on this handle. Returns a freshly-allocated,
432/// NUL-terminated C string that the **caller owns and must free
433/// with `net_free_string`**. Returns NULL when no error has been
434/// recorded.
435///
436/// An owned copy (rather than a borrow into the handle) is
437/// deliberate: a borrowed pointer into the handle's
438/// `Mutex`-owned `CString` would dangle the moment a concurrent
439/// op overwrote the slot or `_free` dropped the inner — a
440/// use-after-free under the multi-threaded usage this handle
441/// advertises. We snapshot under the lock and hand back an
442/// independent allocation, so the returned pointer's lifetime is
443/// the caller's alone.
444#[unsafe(no_mangle)]
445pub unsafe extern "C" fn net_registry_last_error_detail(
446    handle: *mut RegistryClientHandle,
447) -> *mut c_char {
448    if handle.is_null() {
449        return std::ptr::null_mut();
450    }
451    let h: &RegistryClientHandle = unsafe { &*handle };
452    let _op = match h.guard.try_enter() {
453        Some(op) => op,
454        None => return std::ptr::null_mut(),
455    };
456    let guard = h.last_error_detail.lock();
457    match guard.as_ref() {
458        // Clone the contents out from under the lock into a fresh
459        // allocation the caller frees with `net_free_string`.
460        Some(c) => c.clone().into_raw(),
461        None => std::ptr::null_mut(),
462    }
463}
464
465// ─── Visibility setter ───
466
467/// Register a channel with a specific [`Visibility`] tier.
468/// Mirrors `Mesh::register_channel` from the Rust SDK at the C
469/// boundary. `visibility` is an [`i32`] matching the
470/// [`NetVisibility`] discriminants.
471///
472/// Returns `NET_REGISTRY_OK` on success or a typed error code.
473/// Operator-facing detail (e.g. "invalid channel name") is
474/// written to a side-channel: the substrate logs via `tracing`
475/// — no per-call detail string is allocated at this layer.
476#[unsafe(no_mangle)]
477pub unsafe extern "C" fn net_register_channel(
478    mesh_handle: *mut MeshNodeHandle,
479    name: *const c_char,
480    visibility: c_int,
481) -> c_int {
482    if mesh_handle.is_null() || name.is_null() {
483        return NET_REGISTRY_ERR_INVALID_ARGS;
484    }
485    let vis = match NetVisibility::from_raw(visibility) {
486        Some(v) => v,
487        None => return NET_REGISTRY_ERR_INVALID_ARGS,
488    };
489    let name_str = match unsafe { CStr::from_ptr(name).to_str() } {
490        Ok(s) => s,
491        Err(_) => return NET_REGISTRY_ERR_INVALID_ARGS,
492    };
493    let channel = match ChannelName::new(name_str) {
494        Ok(c) => c,
495        Err(_) => return NET_REGISTRY_ERR_INVALID_ARGS,
496    };
497    // Use the mesh's installed ChannelConfigRegistry. The
498    // mesh-FFI's net_mesh_new always installs one, so this is
499    // safe; if it ever changes, the registry being `None` is
500    // surfaced as NET_REGISTRY_ERR_INVALID_ARGS.
501    let Some(mesh_arc) = (unsafe { super::mesh::mesh_node_arc(&*mesh_handle) }) else {
502        return NET_REGISTRY_ERR_INVALID_ARGS;
503    };
504    let Some(configs) = mesh_arc.channel_configs() else {
505        return NET_REGISTRY_ERR_INVALID_ARGS;
506    };
507    let cfg = ChannelConfig::new(ChannelId::new(channel)).with_visibility(vis);
508    configs.insert(cfg);
509    NET_REGISTRY_OK
510}
511
512// ─── FoldQueryClient handle ───
513
514/// FFI handle for a [`FoldQueryClient`]. Same sync model as
515/// [`RegistryClientHandle`]: the inner client lives behind a
516/// `RwLock` so `set_ttl` / `set_deadline` writers serialize with
517/// in-flight ops, and the cache (held by the inner client's
518/// `Arc<RwLock<HashMap<...>>>`) survives deadline / TTL changes.
519///
520/// Same `HandleGuard` quiescing recipe as
521/// [`RegistryClientHandle`].
522pub struct FoldQueryClientHandle {
523    client: ManuallyDrop<ParkingRwLock<FoldQueryClient>>,
524    last_error_detail: ManuallyDrop<ParkingMutex<Option<CString>>>,
525    guard: HandleGuard,
526}
527
528/// Construct a `FoldQueryClient` against an existing
529/// [`MeshNodeHandle`]. Returns a handle the caller frees via
530/// [`net_fold_query_client_free`]. Returns NULL on null input.
531#[unsafe(no_mangle)]
532pub unsafe extern "C" fn net_fold_query_client_new(
533    mesh_handle: *mut MeshNodeHandle,
534) -> *mut FoldQueryClientHandle {
535    if mesh_handle.is_null() {
536        return std::ptr::null_mut();
537    }
538    let Some(mesh_arc) = (unsafe { super::mesh::mesh_node_arc(&*mesh_handle) }) else {
539        return std::ptr::null_mut();
540    };
541    let boxed = Box::new(FoldQueryClientHandle {
542        client: ManuallyDrop::new(ParkingRwLock::new(FoldQueryClient::new(mesh_arc))),
543        last_error_detail: ManuallyDrop::new(ParkingMutex::new(None)),
544        guard: HandleGuard::new(),
545    });
546    Box::into_raw(boxed)
547}
548
549/// Free a `FoldQueryClient` handle. Idempotent on NULL and on a
550/// second call. Quiesces in-flight ops before dropping the inner;
551/// the box stays leaked so a concurrent op can't UAF the handle.
552#[unsafe(no_mangle)]
553pub unsafe extern "C" fn net_fold_query_client_free(handle: *mut FoldQueryClientHandle) {
554    if handle.is_null() {
555        return;
556    }
557    let h: &FoldQueryClientHandle = unsafe { &*handle };
558    match h.guard.begin_free_detailed(FFI_HANDLE_FREE_DEADLINE) {
559        BeginFree::Drained => {
560            // SAFETY: drained; sole writable reference. Box leaked.
561            unsafe {
562                ManuallyDrop::drop(&mut (*handle).client);
563                ManuallyDrop::drop(&mut (*handle).last_error_detail);
564            }
565        }
566        // Benign repeat free — a prior call owns the inner; nothing
567        // to do and nothing leaked by this call.
568        BeginFree::AlreadyFreeing => {}
569        BeginFree::TimedOut => {
570            tracing::warn!(
571                "net_fold_query_client_free: in-flight ops did not drain within deadline; \
572                 leaking inner to avoid use-after-free"
573            );
574        }
575    }
576}
577
578/// Override the cache TTL in milliseconds. `millis == 0` disables
579/// the cache entirely. Mutates in place — the warmed cache
580/// survives the adjustment.
581#[unsafe(no_mangle)]
582pub unsafe extern "C" fn net_fold_query_client_set_ttl(
583    handle: *mut FoldQueryClientHandle,
584    millis: u64,
585) {
586    if handle.is_null() {
587        return;
588    }
589    let h: &FoldQueryClientHandle = unsafe { &*handle };
590    let _op = match h.guard.try_enter() {
591        Some(op) => op,
592        None => return,
593    };
594    h.client.write().set_ttl_mut(Duration::from_millis(millis));
595}
596
597/// Override the per-call deadline in milliseconds. `millis == 0`
598/// resets to the substrate default. Mutates in place.
599#[unsafe(no_mangle)]
600pub unsafe extern "C" fn net_fold_query_client_set_deadline(
601    handle: *mut FoldQueryClientHandle,
602    millis: u64,
603) {
604    if handle.is_null() {
605        return;
606    }
607    let h: &FoldQueryClientHandle = unsafe { &*handle };
608    let _op = match h.guard.try_enter() {
609        Some(op) => op,
610        None => return,
611    };
612    let deadline = if millis == 0 {
613        DEFAULT_QUERY_DEADLINE
614    } else {
615        Duration::from_millis(millis)
616    };
617    h.client.write().set_deadline_mut(deadline);
618}
619
620/// Query the aggregator's latest cached summaries. Cache hit
621/// returns immediately; miss issues a wire RPC, caches the
622/// response, and returns. Returns a JSON-encoded
623/// `[SummaryAnnouncementJson]` string the caller frees via
624/// `net_free_string`.
625#[unsafe(no_mangle)]
626pub unsafe extern "C" fn net_fold_query_client_query_latest(
627    handle: *mut FoldQueryClientHandle,
628    target_node_id: u64,
629    kind: u16,
630    out_error_kind: *mut c_int,
631) -> *mut c_char {
632    if out_error_kind.is_null() {
633        return std::ptr::null_mut();
634    }
635    unsafe {
636        fold_query_op_json(handle, out_error_kind, |client| {
637            block_on(client.query_latest(target_node_id, kind))
638                .map(|summaries| summaries_to_json(&summaries))
639        })
640    }
641}
642
643/// Force a fresh `SummarizeNow` query — never cached.
644#[unsafe(no_mangle)]
645pub unsafe extern "C" fn net_fold_query_client_query_summarize_now(
646    handle: *mut FoldQueryClientHandle,
647    target_node_id: u64,
648    kind: u16,
649    out_error_kind: *mut c_int,
650) -> *mut c_char {
651    if out_error_kind.is_null() {
652        return std::ptr::null_mut();
653    }
654    unsafe {
655        fold_query_op_json(handle, out_error_kind, |client| {
656            block_on(client.query_summarize_now(target_node_id, kind))
657                .map(|summaries| summaries_to_json(&summaries))
658        })
659    }
660}
661
662/// Drop every cached entry.
663#[unsafe(no_mangle)]
664pub unsafe extern "C" fn net_fold_query_client_invalidate_cache(
665    handle: *mut FoldQueryClientHandle,
666) {
667    if handle.is_null() {
668        return;
669    }
670    let h: &FoldQueryClientHandle = unsafe { &*handle };
671    let _op = match h.guard.try_enter() {
672        Some(op) => op,
673        None => return,
674    };
675    h.client.read().invalidate_cache();
676}
677
678/// Drop only cache entries matching `target_node_id`.
679#[unsafe(no_mangle)]
680pub unsafe extern "C" fn net_fold_query_client_invalidate_target(
681    handle: *mut FoldQueryClientHandle,
682    target_node_id: u64,
683) {
684    if handle.is_null() {
685        return;
686    }
687    let h: &FoldQueryClientHandle = unsafe { &*handle };
688    let _op = match h.guard.try_enter() {
689        Some(op) => op,
690        None => return,
691    };
692    h.client.read().invalidate_target(target_node_id);
693}
694
695/// Operator-facing detail string for the most recent non-OK
696/// fold-query op. Returns a freshly-allocated, caller-owned C
697/// string to be freed with `net_free_string` (NULL if no error
698/// recorded). Same owned-copy rationale as
699/// [`net_registry_last_error_detail`] — never a borrow into the
700/// handle, which would dangle on a concurrent overwrite/free.
701#[unsafe(no_mangle)]
702pub unsafe extern "C" fn net_fold_query_last_error_detail(
703    handle: *mut FoldQueryClientHandle,
704) -> *mut c_char {
705    if handle.is_null() {
706        return std::ptr::null_mut();
707    }
708    let h: &FoldQueryClientHandle = unsafe { &*handle };
709    let _op = match h.guard.try_enter() {
710        Some(op) => op,
711        None => return std::ptr::null_mut(),
712    };
713    let guard = h.last_error_detail.lock();
714    match guard.as_ref() {
715        Some(c) => c.clone().into_raw(),
716        None => std::ptr::null_mut(),
717    }
718}
719
720// ─── Internals ───
721
722/// Run a future to completion on the shared mesh-FFI tokio
723/// runtime. Same as `ffi::mesh::block_on` — re-uses that
724/// runtime so we don't fragment scheduling.
725fn block_on<F: std::future::Future>(future: F) -> F::Output {
726    super::mesh::block_on(future)
727}
728
729/// Funnel for any fold-query op that returns a JSON string.
730/// Mirror of [`registry_op_json`].
731unsafe fn fold_query_op_json<F>(
732    handle: *mut FoldQueryClientHandle,
733    out_error_kind: *mut c_int,
734    op: F,
735) -> *mut c_char
736where
737    F: FnOnce(FoldQueryClient) -> Result<String, FoldQueryClientError>,
738{
739    if handle.is_null() {
740        unsafe { write_kind(out_error_kind, NET_REGISTRY_ERR_INVALID_ARGS) };
741        return std::ptr::null_mut();
742    }
743    let h: &FoldQueryClientHandle = unsafe { &*handle };
744    // Guard held only for the clone — see `registry_op_json` for why
745    // the blocking RPC runs unguarded (so a long op deadline can't
746    // make a concurrent `_free` time out and leak the inner).
747    let client = match h.guard.try_enter() {
748        Some(_op) => h.client.read().clone(),
749        None => {
750            unsafe { write_kind(out_error_kind, NET_REGISTRY_ERR_INVALID_ARGS) };
751            return std::ptr::null_mut();
752        }
753    };
754    match op(client) {
755        Ok(json) => unsafe { json_to_raw(json, out_error_kind) },
756        Err(e) => {
757            let (kind, detail) = classify_fold_query(&e);
758            if let Some(_op) = h.guard.try_enter() {
759                store_fold_query_error_detail(h, detail);
760            }
761            unsafe { write_kind(out_error_kind, kind) };
762            std::ptr::null_mut()
763        }
764    }
765}
766
767fn classify_fold_query(err: &FoldQueryClientError) -> (i32, String) {
768    match err {
769        FoldQueryClientError::Transport(e) => (NET_REGISTRY_ERR_TRANSPORT, format!("{e}")),
770        FoldQueryClientError::Codec(c) => (NET_REGISTRY_ERR_CODEC, c.clone()),
771        FoldQueryClientError::Server(FoldQueryError::UnknownKind { kind }) => (
772            NET_REGISTRY_ERR_UNKNOWN_KIND,
773            format!("unknown fold kind: 0x{kind:04x}"),
774        ),
775        FoldQueryClientError::Server(FoldQueryError::DecodeFailed(s)) => {
776            (NET_REGISTRY_ERR_CODEC, format!("server decode: {s}"))
777        }
778    }
779}
780
781fn store_fold_query_error_detail(h: &FoldQueryClientHandle, detail: String) {
782    let c = match CString::new(detail) {
783        Ok(c) => c,
784        Err(_) => CString::new("invalid utf-8 in error detail").unwrap_or_default(),
785    };
786    *h.last_error_detail.lock() = Some(c);
787}
788
789fn summaries_to_json(summaries: &[SummaryAnnouncement]) -> String {
790    let wire: Vec<SummaryWire<'_>> = summaries.iter().map(SummaryWire::from).collect();
791    // `to_string` only fails on serializer-side issues — none of
792    // our wire types have non-string map keys or Float NaN — so
793    // the unwrap is unreachable. Defensive `to_string`-on-error
794    // keeps the FFI surface infallible.
795    serde_json::to_string(&wire).unwrap_or_else(|_| "[]".to_string())
796}
797
798#[cfg(test)]
799fn summary_to_json(s: &SummaryAnnouncement) -> String {
800    serde_json::to_string(&SummaryWire::from(s)).unwrap_or_else(|_| "{}".to_string())
801}
802
803#[derive(serde::Serialize)]
804struct SummaryWire<'a> {
805    fold_kind: u16,
806    source_subnet: String,
807    generation: u64,
808    buckets: Vec<BucketWire<'a>>,
809}
810
811#[derive(serde::Serialize)]
812struct BucketWire<'a> {
813    name: &'a str,
814    count: u64,
815}
816
817impl<'a> From<&'a SummaryAnnouncement> for SummaryWire<'a> {
818    fn from(s: &'a SummaryAnnouncement) -> Self {
819        Self {
820            fold_kind: s.fold_kind,
821            source_subnet: format!("{}", s.source_subnet),
822            generation: s.generation,
823            buckets: s
824                .buckets
825                .iter()
826                .map(|(n, c)| BucketWire {
827                    name: n.as_str(),
828                    count: *c,
829                })
830                .collect(),
831        }
832    }
833}
834
835/// Map a `RegistryClientError` to `(error_kind, detail_string)`.
836fn classify(err: &RegistryClientError) -> (i32, String) {
837    match err {
838        RegistryClientError::Transport(e) => (NET_REGISTRY_ERR_TRANSPORT, format!("{e}")),
839        RegistryClientError::Codec(c) => (NET_REGISTRY_ERR_CODEC, c.clone()),
840        RegistryClientError::Server(RegistryRpcError::DecodeFailed(s)) => {
841            (NET_REGISTRY_ERR_CODEC, format!("server decode: {s}"))
842        }
843        RegistryClientError::Server(RegistryRpcError::UnknownTemplate(t)) => (
844            NET_REGISTRY_ERR_UNKNOWN_TEMPLATE,
845            format!("unknown template: {t}"),
846        ),
847        RegistryClientError::Server(RegistryRpcError::DuplicateGroupName(n)) => (
848            NET_REGISTRY_ERR_DUPLICATE_GROUP_NAME,
849            format!("duplicate group name: {n}"),
850        ),
851        RegistryClientError::Server(RegistryRpcError::SpawnRejected(d)) => (
852            NET_REGISTRY_ERR_SPAWN_REJECTED,
853            format!("spawn rejected: {d}"),
854        ),
855        RegistryClientError::Server(RegistryRpcError::SpawnNotSupported) => (
856            NET_REGISTRY_ERR_SPAWN_NOT_SUPPORTED,
857            "daemon is read-only (no spawn handler installed)".to_string(),
858        ),
859        RegistryClientError::Server(RegistryRpcError::UnknownGroup(g)) => (
860            NET_REGISTRY_ERR_UNKNOWN_GROUP,
861            format!("unknown group: {g}"),
862        ),
863        RegistryClientError::Server(RegistryRpcError::ScaleRejected(d)) => (
864            NET_REGISTRY_ERR_SCALE_REJECTED,
865            format!("scale rejected: {d}"),
866        ),
867        RegistryClientError::Server(RegistryRpcError::ScaleNotSupported) => (
868            NET_REGISTRY_ERR_SCALE_NOT_SUPPORTED,
869            "daemon doesn't accept dynamic scale (no scaler installed)".to_string(),
870        ),
871        RegistryClientError::Server(RegistryRpcError::Unauthorized) => (
872            NET_REGISTRY_ERR_UNAUTHORIZED,
873            "caller is not an operator of the target daemon's aggregator registry".to_string(),
874        ),
875    }
876}
877
878fn store_error_detail(h: &RegistryClientHandle, detail: String) {
879    let c = match CString::new(detail) {
880        Ok(c) => c,
881        Err(_) => CString::new("invalid utf-8 in error detail").unwrap_or_default(),
882    };
883    *h.last_error_detail.lock() = Some(c);
884}
885
886/// Encode the wire-contract JSON for a slice of registry-group
887/// summaries via `serde_json`. The substrate types serialize their
888/// byte arrays as arrays of u8; the proxy wire-types below render them
889/// as hex strings instead.
890///
891/// **Breaking:** `group_seed_hex` (64 chars, the raw seed) is replaced
892/// by `group_seed_fingerprint_hex` (16 chars). The seed
893/// deterministically derives every replica keypair, and a status API
894/// should not carry private key material — see
895/// `RegistryGroupSummary::group_seed_fingerprint`. Consumers using the
896/// old field to correlate groups should use the fingerprint;
897/// consumers using it to *derive replica keys* were relying on
898/// something that is not an authorization primitive and is often
899/// recomputable from the group name anyway.
900fn groups_to_json(groups: &[RegistryGroupSummary]) -> String {
901    let wire: Vec<GroupWire<'_>> = groups.iter().map(GroupWire::from).collect();
902    serde_json::to_string(&wire).unwrap_or_else(|_| "[]".to_string())
903}
904
905fn group_to_json(g: &RegistryGroupSummary) -> String {
906    serde_json::to_string(&GroupWire::from(g)).unwrap_or_else(|_| "{}".to_string())
907}
908
909#[derive(serde::Serialize)]
910struct GroupWire<'a> {
911    name: &'a str,
912    group_seed_fingerprint_hex: String,
913    replicas: Vec<ReplicaWire<'a>>,
914}
915
916#[derive(serde::Serialize)]
917struct ReplicaWire<'a> {
918    generation: u64,
919    healthy: bool,
920    diagnostic: Option<&'a str>,
921    placement_node_id: Option<u64>,
922}
923
924impl<'a> From<&'a RegistryGroupSummary> for GroupWire<'a> {
925    fn from(g: &'a RegistryGroupSummary) -> Self {
926        Self {
927            name: g.name.as_str(),
928            group_seed_fingerprint_hex: g.group_seed_fingerprint.to_hex(),
929            replicas: g
930                .replicas
931                .iter()
932                .map(|r| ReplicaWire {
933                    generation: r.generation,
934                    healthy: r.healthy,
935                    diagnostic: r.diagnostic.as_deref(),
936                    placement_node_id: r.placement_node_id,
937                })
938                .collect(),
939        }
940    }
941}
942
943#[cfg(test)]
944mod tests {
945    use super::*;
946
947    #[test]
948    fn visibility_round_trips_through_raw() {
949        for (raw, expected) in [
950            (0, Visibility::Global),
951            (1, Visibility::ParentVisible),
952            (2, Visibility::Exported),
953            (3, Visibility::SubnetLocal),
954        ] {
955            let back = NetVisibility::from_raw(raw).expect("known discriminant");
956            assert_eq!(format!("{back:?}"), format!("{expected:?}"));
957        }
958        assert!(NetVisibility::from_raw(99).is_none());
959        assert!(NetVisibility::from_raw(-1).is_none());
960    }
961
962    #[test]
963    fn group_to_json_includes_every_documented_field() {
964        let g = RegistryGroupSummary {
965            name: "alpha".into(),
966            group_seed_fingerprint: crate::adapter::net::behavior::aggregator::SeedFingerprint::of(
967                &[0xABu8; 32],
968            ),
969            source_subnet: crate::adapter::net::subnet::SubnetId::GLOBAL,
970            fold_kinds: vec![0x0001],
971            replicas: vec![
972                crate::adapter::net::behavior::aggregator::RegistryReplicaSummary {
973                    generation: 42,
974                    healthy: true,
975                    diagnostic: None,
976                    placement_node_id: Some(0xBEEF),
977                },
978                crate::adapter::net::behavior::aggregator::RegistryReplicaSummary {
979                    generation: 0,
980                    healthy: false,
981                    diagnostic: Some("stuck".into()),
982                    placement_node_id: None,
983                },
984            ],
985        };
986        let json = group_to_json(&g);
987        assert!(json.contains("\"name\":\"alpha\""));
988        // The seed itself must never appear: 32 bytes of 0xAB would
989        // render as 64 alternating "ab" chars.
990        let raw_seed_hex = "ab".repeat(32);
991        assert!(
992            !json.contains(&raw_seed_hex),
993            "the raw group seed was rendered into the wire JSON: {json}"
994        );
995        // The fingerprint is present, 16 hex chars, and is NOT a
996        // prefix of the seed rendering.
997        let fp =
998            crate::adapter::net::behavior::aggregator::SeedFingerprint::of(&[0xABu8; 32]).to_hex();
999        assert_eq!(fp.len(), 16);
1000        assert!(
1001            json.contains(&format!("\"group_seed_fingerprint_hex\":\"{fp}\"")),
1002            "missing group_seed_fingerprint_hex in {json}"
1003        );
1004        assert!(json.contains("\"generation\":42"));
1005        assert!(json.contains("\"healthy\":true"));
1006        assert!(json.contains("\"diagnostic\":null"));
1007        assert!(json.contains("\"placement_node_id\":48879"));
1008        assert!(json.contains("\"healthy\":false"));
1009        assert!(json.contains("\"diagnostic\":\"stuck\""));
1010        assert!(json.contains("\"placement_node_id\":null"));
1011    }
1012
1013    #[test]
1014    fn summary_to_json_includes_every_documented_field() {
1015        let s = SummaryAnnouncement {
1016            fold_kind: 0x42,
1017            source_subnet: crate::adapter::net::subnet::SubnetId::GLOBAL,
1018            generation: 7,
1019            buckets: vec![("alpha".into(), 1), ("beta".into(), 2)],
1020        };
1021        let json = summary_to_json(&s);
1022        assert!(json.contains("\"fold_kind\":66"));
1023        assert!(json.contains("\"source_subnet\":\"global\""));
1024        assert!(json.contains("\"generation\":7"));
1025        assert!(json.contains("\"name\":\"alpha\""));
1026        assert!(json.contains("\"count\":1"));
1027        assert!(json.contains("\"name\":\"beta\""));
1028        assert!(json.contains("\"count\":2"));
1029    }
1030
1031    #[test]
1032    fn classify_fold_query_maps_every_variant() {
1033        use crate::adapter::net::mesh_rpc::RpcError;
1034        // Transport — anything carrying an RpcError lands on
1035        // NET_REGISTRY_ERR_TRANSPORT regardless of the inner kind.
1036        let transport = FoldQueryClientError::Transport(RpcError::NoRoute {
1037            target: 0,
1038            reason: String::new(),
1039        });
1040        assert_eq!(
1041            classify_fold_query(&transport).0,
1042            NET_REGISTRY_ERR_TRANSPORT
1043        );
1044
1045        let codec = FoldQueryClientError::Codec("bad".into());
1046        assert_eq!(classify_fold_query(&codec).0, NET_REGISTRY_ERR_CODEC);
1047
1048        let unknown_kind = FoldQueryClientError::Server(FoldQueryError::UnknownKind { kind: 0x42 });
1049        let (kind_code, detail) = classify_fold_query(&unknown_kind);
1050        assert_eq!(kind_code, NET_REGISTRY_ERR_UNKNOWN_KIND);
1051        assert!(detail.contains("0x0042"));
1052
1053        let decode_failed =
1054            FoldQueryClientError::Server(FoldQueryError::DecodeFailed("boom".into()));
1055        assert_eq!(
1056            classify_fold_query(&decode_failed).0,
1057            NET_REGISTRY_ERR_CODEC,
1058        );
1059    }
1060}