Merge upstream develop

Author: github-actions[bot]

packages/react-native/src/create-polycentric-client.shared.ts

@@ -10,6 +10,55 @@ export function normalizeDatabaseName(databaseName?: string) {
   return (databaseName ?? 'polycentric').trim() || 'polycentric';
 }
 
+/**
+ * Build a non-blocking log sink for `setLogger`.
+ *
+ * Every rs-core log crosses the FFI boundary as a synchronous host call.
+ * Calling `console.log` inline for each one stalls the JS thread under a
+ * burst (RN's `console.log` is slow, worse with a debugger attached), which
+ * can freeze the UI. Instead we enqueue messages and flush them in a single
+ * batched `console.log` on the next tick, and drop oldest under backpressure
+ * so a flood can neither block the caller nor grow unbounded.
+ */
+export function createBatchingLogSink(maxQueue = 1000): {
+  log: (message: string) => void;
+} {
+  const queue: string[] = [];
+  let scheduled = false;
+  let dropped = 0;
+
+  const flush = () => {
+    scheduled = false;
+    if (dropped > 0) {
+      // eslint-disable-next-line no-console
+      console.warn(
+        `[polycentric] dropped ${dropped} log message(s) (backpressure)`,
+      );
+      dropped = 0;
+    }
+    if (queue.length === 0) return;
+    const batch = queue.splice(0, queue.length).join('\n');
+    // eslint-disable-next-line no-console
+    console.log(batch);
+  };
+
+  return {
+    log: (message: string) => {
+      if (queue.length >= maxQueue) {
+        dropped++;
+        return;
+      }
+      queue.push(message);
+      if (!scheduled) {
+        scheduled = true;
+        // Defer off the FFI caller so the Rust->JS callback returns
+        // immediately; many logs in one tick collapse into one console.log.
+        setTimeout(flush, 0);
+      }
+    },
+  };
+}
+
 /**
  * Publishes a new Identity document authorized by the client's current
  * keypair, and registers `server` so the new identity is synced there.

packages/react-native/src/create-polycentric-client.ts

@@ -4,6 +4,7 @@ import { createReactNativeStorageDriver } from './datastore/expo-sqlite';
 import { createReactNativeFileStoreDriver } from './filestore/expo-file-system';
 import { ReactNativeCryptoManager } from './crypto/react-native-crypto-manager';
 import {
+  createBatchingLogSink,
   createIdentity,
   normalizeDatabaseName,
   type CreatePolycentricClientConfig,
@@ -13,11 +14,7 @@ let loggerInstalled = false;
 function installConsoleLogger() {
   if (loggerInstalled) return;
   loggerInstalled = true;
-  setLogger({
-    log: (message: string) => {
-      console.log(message);
-    },
-  });
+  setLogger(createBatchingLogSink());
 }
 
 export async function createPolycentricClient(

packages/react-native/src/create-polycentric-client.web.ts

@@ -10,6 +10,7 @@ import {
   uniffiInitAsync,
 } from '@polycentric/rs-core-uniffi-web';
 import {
+  createBatchingLogSink,
   createIdentity,
   normalizeDatabaseName,
   type CreatePolycentricClientConfig,
@@ -19,11 +20,7 @@ let loggerInstalled = false;
 function installConsoleLogger() {
   if (loggerInstalled) return;
   loggerInstalled = true;
-  setLogger({
-    log: (message: string) => {
-      console.log(message);
-    },
-  });
+  setLogger(createBatchingLogSink());
 }
 
 export async function createPolycentricClient(

packages/rs-core/src/logging.rs

@@ -1,30 +1,97 @@
 //! Foreign-implemented log sink. Hosts (JS, Kotlin, Swift) register
-//! an implementation via `set_logger`; rs-core code that calls
-//! `log_msg` forwards messages to it. Cross-target by construction —
-//! native targets and wasm both reach the host through the same FFI
-//! callback uniffi already builds for us.
+//! an implementation via `set_logger`; rs-core code logs through the
+//! leveled helpers below. Cross-target by construction — native targets
+//! and wasm both reach the host through the same FFI callback uniffi
+//! already builds for us.
+//!
+//! Every message crosses the FFI boundary as a synchronous host call, so a
+//! burst of messages can stall the host's thread (on React Native, each call
+//! marshals across the bridge into a `console.log`). Two guards keep that in
+//! check:
+//!   * **Level filtering** — messages below the host-configured threshold
+//!     ([`set_log_level`], default [`LogLevel::Info`]) are dropped in Rust and
+//!     never cross FFI. The message string isn't even formatted (the helpers
+//!     take a closure), so filtered logging is nearly free.
+//!   * Hosts are still expected to make their sink non-blocking (batch/drop)
+//!     for defense in depth.
 
+use std::sync::atomic::{AtomicU8, Ordering};
 use std::sync::{Arc, Mutex};
 
+/// Severity of a log message. Hosts set a minimum threshold via
+/// [`set_log_level`]; anything below it is dropped before crossing FFI.
+#[derive(Clone, Copy, PartialEq, Eq, Debug, uniffi::Enum)]
+pub enum LogLevel {
+    Trace,
+    Debug,
+    Info,
+    Warn,
+    Error,
+    /// Disables all logging.
+    Off,
+}
+
+impl LogLevel {
+    fn rank(self) -> u8 {
+        match self {
+            LogLevel::Trace => 0,
+            LogLevel::Debug => 1,
+            LogLevel::Info => 2,
+            LogLevel::Warn => 3,
+            LogLevel::Error => 4,
+            LogLevel::Off => 5,
+        }
+    }
+}
+
 #[uniffi::export(with_foreign)]
 pub trait Logger: Send + Sync {
     fn log(&self, message: String);
 }
 
 static LOGGER: Mutex<Option<Arc<dyn Logger>>> = Mutex::new(None);
+/// Minimum level that crosses FFI. Defaults to `Info` so debug/trace
+/// floods are dropped unless a host explicitly opts in.
+static MIN_LEVEL: AtomicU8 = AtomicU8::new(2);
 
 /// Register the foreign logger. Replaces any previously-set value.
 #[uniffi::export]
 pub fn set_logger(logger: Arc<dyn Logger>) {
     *LOGGER.lock().unwrap() = Some(logger);
 }
 
-/// Forward `message` to the registered foreign logger (if any). The
-/// Arc is cloned out of the mutex before the foreign call so a
-/// re-entrant logger impl can't deadlock against `set_logger`.
-pub(crate) fn log_msg(message: impl Into<String>) {
+/// Set the minimum level forwarded to the host. Messages below this are
+/// dropped in Rust without crossing the FFI boundary.
+#[uniffi::export]
+pub fn set_log_level(level: LogLevel) {
+    MIN_LEVEL.store(level.rank(), Ordering::Relaxed);
+}
+
+/// Log at `level`. The message is only built (and only crosses FFI) when
+/// `level` is at or above the configured threshold — pass a closure so the
+/// `format!` cost is skipped for filtered messages.
+pub(crate) fn log_at(level: LogLevel, message: impl FnOnce() -> String) {
+    if level.rank() < MIN_LEVEL.load(Ordering::Relaxed) {
+        return;
+    }
+    // Clone the Arc out of the mutex before the foreign call so a
+    // re-entrant logger impl can't deadlock against `set_logger`.
     let logger = LOGGER.lock().unwrap().clone();
     if let Some(l) = logger {
-        l.log(message.into());
+        l.log(message());
     }
 }
+
+pub(crate) fn log_debug(message: impl FnOnce() -> String) {
+    log_at(LogLevel::Debug, message);
+}
+
+#[allow(dead_code)]
+pub(crate) fn log_info(message: impl FnOnce() -> String) {
+    log_at(LogLevel::Info, message);
+}
+
+#[allow(dead_code)]
+pub(crate) fn log_warn(message: impl FnOnce() -> String) {
+    log_at(LogLevel::Warn, message);
+}

packages/rs-core/src/query/event.rs

@@ -135,6 +135,7 @@ fn merge_event(
     client: &std::sync::Arc<std::sync::Mutex<crate::client::PolycentricClient>>,
 ) -> Vec<u8> {
     let c = client.lock().unwrap();
+    let mut first_error: Option<String> = None;
     for v in values {
         if v.is_empty() {
             continue;
@@ -148,10 +149,15 @@ fn merge_event(
         match c.validate_event(signed, &bundle.event_proofs) {
             Ok(()) => return v.clone(),
             Err(e) => {
-                crate::logging::log_msg(format!("[merge_event] dropping bundle: {e:?}"));
+                first_error.get_or_insert_with(|| format!("{e:?}"));
             }
         }
     }
+    if let Some(reason) = first_error {
+        crate::logging::log_debug(|| {
+            format!("[merge_event] no valid bundle; first reason: {reason}")
+        });
+    }
     Vec::new()
 }
 

packages/rs-core/src/query/validation.rs

@@ -1,38 +1,64 @@
 //! Shared validation passes for per-RPC merge functions. Each merge
 //! function combines per-server responses and then runs these to drop
 //! any bundles/hints that don't pass `PolycentricClient::validate_event`.
+//!
+//! Dropping is expected and routine — e.g. an identity whose chain can't be
+//! fully reconstructed (a deleted or not-yet-synced identity event leaves a
+//! gap) makes every later event fail validation. To avoid flooding the log
+//! (and freezing the UI) we emit at most one aggregated line per pass rather
+//! than one per dropped item.
 
 use polycentric_common::models::protos_v2::{EventBundle, EventHint};
 
 use crate::client::PolycentricClient;
 
 /// Retain only bundles whose signed event passes `validate_event`.
 pub(crate) fn retain_validated_bundles(client: &PolycentricClient, bundles: &mut Vec<EventBundle>) {
+    let before = bundles.len();
+    let mut first_error: Option<String> = None;
     bundles.retain(|b| match b.signed_event.as_ref() {
         Some(se) => match client.validate_event(se, &b.event_proofs) {
             Ok(()) => true,
             Err(e) => {
-                crate::logging::log_msg(format!("[merge] dropping bundle: {e:?}"));
+                first_error.get_or_insert_with(|| format!("{e:?}"));
                 false
             }
         },
         None => false,
     });
+    log_dropped("bundle", before - bundles.len(), first_error);
 }
 
 /// Retain only hints whose signed event passes `validate_event`.
 pub(crate) fn retain_validated_hints(client: &PolycentricClient, hints: &mut Vec<EventHint>) {
-    hints.retain(|h| match h.event_bundle.as_ref() {
-        Some(b) => match b.signed_event.as_ref() {
-            Some(se) => match client.validate_event(se, &b.event_proofs) {
+    let before = hints.len();
+    let mut first_error: Option<String> = None;
+    hints.retain(|h| {
+        match h
+            .event_bundle
+            .as_ref()
+            .and_then(|b| b.signed_event.as_ref().map(|se| (b, se)))
+        {
+            Some((b, se)) => match client.validate_event(se, &b.event_proofs) {
                 Ok(()) => true,
                 Err(e) => {
-                    crate::logging::log_msg(format!("[merge] dropping hint: {e:?}"));
+                    first_error.get_or_insert_with(|| format!("{e:?}"));
                     false
                 }
             },
             None => false,
-        },
-        None => false,
+        }
+    });
+    log_dropped("hint", before - hints.len(), first_error);
+}
+
+/// Emit a single aggregated line when a pass dropped anything.
+fn log_dropped(kind: &str, count: usize, first_error: Option<String>) {
+    if count == 0 {
+        return;
+    }
+    let reason = first_error.unwrap_or_else(|| "missing signed event".to_string());
+    crate::logging::log_debug(|| {
+        format!("[merge] dropped {count} unvalidated {kind}(s); first reason: {reason}")
     });
 }