diff --git a/CHANGELOG.md b/CHANGELOG.md index 1cb143ed..aaf33d7e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,7 @@ - CocoaPods support removed. - Minimum Flutter version is now 3.47.1 due to improved build hooks support - Completely remove 'rust-android-gradle' because it is failing to keep up with the rest of the ecosystem (AGP, Kotlin, etc) + - Fix multiple-attach idempotency - veilid-python: - More tests diff --git a/veilid-core/smoketest/Cargo.lock b/veilid-core/smoketest/Cargo.lock index 5932b2c0..32f95942 100644 --- a/veilid-core/smoketest/Cargo.lock +++ b/veilid-core/smoketest/Cargo.lock @@ -1515,9 +1515,9 @@ checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" [[package]] name = "hermit-abi" -version = "0.5.2" +version = "0.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" +checksum = "e17592d60ebacc7d5e169f4663c5f84f9161cc90328abcfe8456f41e4dfcb284" [[package]] name = "hex" @@ -3542,9 +3542,9 @@ checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" [[package]] name = "value-bag" -version = "1.13.2" +version = "1.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "068e763e8279de7ab94b6afebded2cb701678af094feb1c12ccb061b4783c1be" +checksum = "11d4788a546b899f8656382fe210e0c478fe3d2dcd6263d50cb6cc268240de6c" [[package]] name = "vcpkg" diff --git a/veilid-core/src/protected_store/mod.rs b/veilid-core/src/protected_store/mod.rs index 55cb97e8..1acfa00d 100644 --- a/veilid-core/src/protected_store/mod.rs +++ b/veilid-core/src/protected_store/mod.rs @@ -217,6 +217,11 @@ impl ProtectedStore { } /// Load a byte buffer secret by key, or `None` if no value is stored for it. + /// + /// `None` means the backend established that nothing is stored. A backend that could not read + /// the store returns [VeilidAPIError::TryAgain] instead (an iOS keychain before first unlock), + /// and the stored value is intact. A caller that reprovisions on `None` must propagate that + /// error rather than fold it into absence, or it overwrites a secret it merely could not read. pub async fn load_user_secret>( &self, context: ProtectedStoreContext, diff --git a/veilid-core/src/protected_store/native.rs b/veilid-core/src/protected_store/native.rs index 6241d553..218203e2 100644 --- a/veilid-core/src/protected_store/native.rs +++ b/veilid-core/src/protected_store/native.rs @@ -30,6 +30,12 @@ impl ProtectedStoreBackend { } } +/// A keyring reported that the store could not be read or written because the device is locked. +/// The stored value is intact, so this is transient and never absence. +fn locked_error() -> VeilidAPIError { + VeilidAPIError::try_again("protected store is locked, retry once the device is unlocked") +} + /// Protected store backend backed by the OS keyring (or an on-disk insecure keyring). /// /// Keyring calls can block on user input (e.g. a keychain permission dialog), so each runs on a @@ -48,6 +54,7 @@ impl KeyringBackend { match km.lock().with_keyring(&service, &key, |kr| kr.get_value()) { Ok(v) => Ok(Some(Zeroizing::new(v))), Err(KeyringError::NoPasswordFound) => Ok(None), + Err(KeyringError::Locked) => Err(locked_error()), Err(e) => Err(VeilidAPIError::generic(format!( "failed to load user secret: {}", e @@ -63,13 +70,18 @@ impl KeyringBackend { let key = key.to_owned(); let value = Zeroizing::new(value.to_owned()); blocking_yielding(move || { - km.lock() - .with_keyring(&service, &key, |kr| { - let existed = kr.get_value().is_ok(); - kr.set_value(value.as_str())?; - Ok(existed) - }) - .map_err(|e| VeilidAPIError::generic(format!("failed to save user secret: {}", e))) + match km.lock().with_keyring(&service, &key, |kr| { + let existed = kr.get_value().is_ok(); + kr.set_value(value.as_str())?; + Ok(existed) + }) { + Ok(existed) => Ok(existed), + Err(KeyringError::Locked) => Err(locked_error()), + Err(e) => Err(VeilidAPIError::generic(format!( + "failed to save user secret: {}", + e + ))), + } }) .await } @@ -85,6 +97,7 @@ impl KeyringBackend { { Ok(_) => Ok(true), Err(KeyringError::NoPasswordFound) => Ok(false), + Err(KeyringError::Locked) => Err(locked_error()), Err(e) => Err(VeilidAPIError::generic(format!( "failed to remove user secret: {}", e @@ -102,6 +115,7 @@ impl KeyringBackend { Err(KeyringError::Unsupported) => Err(VeilidAPIError::generic( "this protected store backend does not support listing keys", )), + Err(KeyringError::Locked) => Err(locked_error()), Err(e) => Err(VeilidAPIError::generic(format!( "failed to list user secrets: {}", e @@ -119,8 +133,6 @@ fn service_name(config: &VeilidConfig) -> String { } } -// async to match the wasm build_backend, which awaits opening its database -#[allow(clippy::unused_async)] pub(super) async fn build_backend(ps: &ProtectedStore) -> VeilidAPIResult { let config = ps.config(); @@ -139,15 +151,24 @@ pub(super) async fn build_backend(ps: &ProtectedStore) -> VeilidAPIResult = None; if !config.protected_store.always_use_insecure_storage { - cfg_if! { - if #[cfg(target_os = "android")] { - let maybe_km = KeyringManager::new_secure(&config.program_name, crate::veilid_api::android::get_android_globals()); - } else { - let maybe_km = KeyringManager::new_secure(&config.program_name); + // Construction probes the store and can block on user input (an unlock dialog), so it + // runs off the async threads like every other keyring call + let program_name = config.program_name.clone(); + let maybe_km = blocking_yielding(move || { + cfg_if! { + if #[cfg(target_os = "android")] { + KeyringManager::new_secure(&program_name, crate::veilid_api::android::get_android_globals()) + } else { + KeyringManager::new_secure(&program_name) + } } - } + }) + .await; keyring_manager = match maybe_km { Ok(v) => Some(v), + // A locked store is not a missing store: falling back to insecure storage would + // read the DEK as absent and re-key over data the locked store still holds + Err(KeyringError::Locked) => return Err(locked_error()), Err(e) => { veilid_log!(ps info "Secure key storage service unavailable, falling back to direct disk-based storage: {}", e); None diff --git a/veilid-core/src/veilid_api/android/mod.rs b/veilid-core/src/veilid_api/android/mod.rs index 7f4cd7fe..e3cc1d53 100644 --- a/veilid-core/src/veilid_api/android/mod.rs +++ b/veilid-core/src/veilid_api/android/mod.rs @@ -28,24 +28,42 @@ lazy_static! { } /// Register the host Android application's JNI environment and `Context` with -/// veilid-core. Must be called once from the app (via JNI) before startup. +/// veilid-core. Must be called from the app (via JNI) before startup. +/// +/// Idempotent. Every Flutter engine creation attaches the plugin and calls this again, including +/// background engines a running node knows nothing about, so a call made with globals already +/// registered checks the incoming `Context` and returns without replacing them. +/// +/// A `Context` that is not the registered one is ignored, not adopted. The expected argument is the +/// process-wide application `Context`, which is a per-process singleton, so a mismatch means the +/// caller passed something narrower and shorter-lived, and swapping it would hand a running node's +/// JNI calls a `Context` that can be torn down under it. pub fn veilid_core_setup_android(mut env: EnvUnowned, ctx: JObject) { env.with_env(|env| -> JniResult<()> { + let mut globals_locked = ANDROID_GLOBALS.lock(); + if let Some(globals) = globals_locked.as_ref() { + if !env.is_same_object(&ctx, globals.ctx.as_obj())? { + warn!("veilid_core_setup_android called with a different Context, keeping the registered one"); + } + return Ok(()); + } let ctx = env.new_global_ref(ctx)?; let vm = env.get_java_vm()?; - *ANDROID_GLOBALS.lock() = Some(AndroidGlobals { vm, ctx }); + *globals_locked = Some(AndroidGlobals { vm, ctx }); Ok(()) }) .resolve::(); } /// Whether [veilid_core_setup_android] has run and the Android globals are set. +#[must_use] pub fn is_android_ready() -> bool { ANDROID_GLOBALS.lock().is_some() } /// Get a thread-attached copy of the Android JavaVM and `Context` globals. /// Panics if [veilid_core_setup_android] has not been called. +#[must_use] pub fn get_android_globals() -> (JavaVM, Global>) { let globals_locked = ANDROID_GLOBALS.lock(); let globals = globals_locked.as_ref().unwrap_or_log(); diff --git a/veilid-flutter/example/dart_test.yaml b/veilid-flutter/example/dart_test.yaml index e482d482..0dc1956b 100644 --- a/veilid-flutter/example/dart_test.yaml +++ b/veilid-flutter/example/dart_test.yaml @@ -4,6 +4,7 @@ concurrency: 1 tags: config: {} + startup: {} crypto: {} table_db: {} routing_context: {} diff --git a/veilid-flutter/example/integration_test/app_test.dart b/veilid-flutter/example/integration_test/app_test.dart index 07df9cb7..be0d2fe8 100644 --- a/veilid-flutter/example/integration_test/app_test.dart +++ b/veilid-flutter/example/integration_test/app_test.dart @@ -14,6 +14,7 @@ import 'test_dht_transactions.dart'; import 'test_dht_transactions_full.dart'; import 'test_protected_store.dart'; import 'test_routing_context.dart'; +import 'test_startup_shutdown.dart'; import 'test_table_db.dart'; import 'test_veilid_config.dart'; @@ -24,6 +25,7 @@ void main() => runWithIntegrationLog(() { final fixture = DefaultVeilidFixture( programName: 'veilid_flutter integration test', logFixture: LogFixture.instance); + final testStartupShutdown = TestStartupShutdown(); final updateProcessorFixture = UpdateProcessorFixture(veilidFixture: fixture); @@ -236,4 +238,30 @@ void main() => runWithIntegrationLog(() { } }); }); + + // Owns its own node lifecycle, so it runs after the fixture has released + // the shared one + group('Startup and Shutdown', () { + setUpAll(testStartupShutdown.setUpAll); + tearDown(testStartupShutdown.tearDown); + + test('two startups then two shutdowns keep the node alive until the last', + testStartupShutdown.testTwoStartupsOneShutdownKeepsNodeAlive, + timeout: const Timeout(Duration(seconds: 120)), tags: ['startup']); + test('joining caller receives updates on its own stream', + testStartupShutdown.testJoinedCallerReceivesUpdates, + timeout: const Timeout(Duration(seconds: 120)), tags: ['startup']); + test('failed startup takes no reference', + testStartupShutdown.testFailedStartupTakesNoReference, + timeout: const Timeout(Duration(seconds: 120)), tags: ['startup']); + test('startup with a different init key is rejected', + testStartupShutdown.testDifferentInitKeyIsRejected, + timeout: const Timeout(Duration(seconds: 120)), tags: ['startup']); + test('extra shutdowns neither underflow nor shut down twice', + testStartupShutdown.testExtraShutdownsDoNotUnderflow, + timeout: const Timeout(Duration(seconds: 120)), tags: ['startup']); + test('a reaped sink holds no reference and leaves the node running', + testStartupShutdown.testReapedSinkHoldsNoReference, + timeout: const Timeout(Duration(seconds: 120)), tags: ['startup']); + }); }); diff --git a/veilid-flutter/example/integration_test/test_startup_shutdown.dart b/veilid-flutter/example/integration_test/test_startup_shutdown.dart new file mode 100644 index 00000000..31495beb --- /dev/null +++ b/veilid-flutter/example/integration_test/test_startup_shutdown.dart @@ -0,0 +1,245 @@ +// Allow environment variables +// ignore_for_file: do_not_use_environment + +import 'dart:async'; +import 'dart:io' show Platform; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:path/path.dart' as p; +import 'package:veilid/veilid.dart'; +import 'package:veilid_test/veilid_test.dart'; + +const _programName = 'veilid_flutter startup test'; +const _updateTimeout = Duration(seconds: 60); + +Future _testConfig(String programName) async { + var config = await getDefaultVeilidConfig( + isWeb: kIsWeb, + programName: programName, + namespace: const String.fromEnvironment('NAMESPACE'), + bootstrap: const String.fromEnvironment('BOOTSTRAP'), + bootstrapKeys: const bool.hasEnvironment('BOOTSTRAP_KEYS') + ? const String.fromEnvironment('BOOTSTRAP_KEYS') + : null, + networkKeyPassword: const String.fromEnvironment('NETWORK_KEY')); + + if (!kIsWeb) { + config = config.copyWith( + tableStore: config.tableStore.copyWith( + directory: p.join(config.tableStore.directory, programName), + delete: true), + protectedStore: config.protectedStore.copyWith( + directory: p.join(config.protectedStore.directory, programName), + delete: true), + blockStore: config.blockStore.copyWith( + directory: p.join(config.blockStore.directory, programName), + delete: true)); + } else { + config = config.copyWith( + tableStore: config.tableStore.copyWith(delete: true), + protectedStore: config.protectedStore.copyWith(delete: true), + blockStore: config.blockStore.copyWith(delete: true)); + } + + return config.copyWith( + capabilities: + // XXX: Remove after https://gitlab.com/veilid/veilid/-/issues/492 + const VeilidConfigCapabilities(disable: ['DHTV']), + protectedStore: config.protectedStore.copyWith( + // Linux often does not have a secret storage mechanism installed + allowInsecureFallback: !kIsWeb && Platform.isLinux)); +} + +/// Keeps an update stream subscribed for as long as its reference is held. +/// +/// Cancelling a subscription drops its sink, so a test that wants to release a +/// reference through shutdownVeilidCore must not cancel first. +class _UpdateCollector { + _UpdateCollector(Stream stream) { + _subscription = stream.listen((update) { + if (update is VeilidUpdateAttachment && !_attachment.isCompleted) { + _attachment.complete(update); + } + if (_pending != null && --_pendingCount <= 0) { + _pending!.complete(); + _pending = null; + } + }, onDone: _done.complete); + } + + final _done = Completer(); + final _attachment = Completer(); + late final StreamSubscription _subscription; + Completer? _pending; + int _pendingCount = 0; + + Future get done => _done.future; + Future get firstAttachment => _attachment.future; + bool get isDone => _done.isCompleted; + + /// Completes once [count] more updates have arrived. + Future nextUpdates(int count) { + final pending = _pending = Completer(); + _pendingCount = count; + return pending.future; + } + + /// The generator only resumes to run its cleanup when the next update + /// arrives, so this hangs unless updates are already flowing. + Future cancel() => _subscription.cancel(); +} + +class TestStartupShutdown { + late final VeilidConfig config; + late final VeilidConfig otherConfig; + + Future setUpAll() async { + config = await _testConfig(_programName); + otherConfig = await _testConfig('$_programName other'); + } + + /// Release anything a failed test left running, so the next test starts from + /// no node at all. + Future tearDown() async { + for (var i = 0; i < 8 && !await Veilid.instance.isShutdown(); i++) { + try { + await Veilid.instance.shutdownVeilidCore(); + } on VeilidAPIExceptionNotInitialized { + break; + } + } + expect(await Veilid.instance.isShutdown(), isTrue); + } + + Future testTwoStartupsOneShutdownKeepsNodeAlive() async { + final firstStream = await Veilid.instance.startupVeilidCore(config); + final secondStream = await Veilid.instance.startupVeilidCore(config); + final first = _UpdateCollector(firstStream); + final second = _UpdateCollector(secondStream); + + expect(await Veilid.instance.isShutdown(), isFalse); + + // retires the second reference only + await Veilid.instance.shutdownVeilidCore(); + expect(await Veilid.instance.isShutdown(), isFalse); + await second.done.timeout(_updateTimeout); + expect(first.isDone, isFalse); + + // the node is still usable through the surviving reference + final state = await Veilid.instance.getVeilidState(); + expect(state.attachment.state, equals(AttachmentState.detached)); + + await Veilid.instance.shutdownVeilidCore(); + expect(await Veilid.instance.isShutdown(), isTrue); + await first.done.timeout(_updateTimeout); + } + + Future testJoinedCallerReceivesUpdates() async { + final firstStream = await Veilid.instance.startupVeilidCore(config); + final secondStream = await Veilid.instance.startupVeilidCore(config); + final first = _UpdateCollector(firstStream); + final second = _UpdateCollector(secondStream); + + await Veilid.instance.attach(); + // each joined caller attaches the node it shares + await Veilid.instance.attach(); + expect(await first.firstAttachment.timeout(_updateTimeout), + isA()); + expect(await second.firstAttachment.timeout(_updateTimeout), + isA()); + await Veilid.instance.detach(); + await Veilid.instance.detach(); + + await Veilid.instance.shutdownVeilidCore(); + await Veilid.instance.shutdownVeilidCore(); + await first.done.timeout(_updateTimeout); + await second.done.timeout(_updateTimeout); + } + + Future testFailedStartupTakesNoReference() async { + final invalidConfig = config.copyWith(programName: '$_programName/invalid'); + await expectLater(() => Veilid.instance.startupVeilidCore(invalidConfig), + throwsA(isA())); + + // one reference, so one shutdown must take the node down + final stream = await Veilid.instance.startupVeilidCore(config); + final collector = _UpdateCollector(stream); + await Veilid.instance.shutdownVeilidCore(); + expect(await Veilid.instance.isShutdown(), isTrue); + await collector.done.timeout(_updateTimeout); + } + + Future testDifferentInitKeyIsRejected() async { + final stream = await Veilid.instance.startupVeilidCore(config); + final collector = _UpdateCollector(stream); + + await expectLater(() => Veilid.instance.startupVeilidCore(otherConfig), + throwsA(isA())); + expect(collector.isDone, isFalse); + + // the rejected caller took no reference + await Veilid.instance.shutdownVeilidCore(); + expect(await Veilid.instance.isShutdown(), isTrue); + await collector.done.timeout(_updateTimeout); + } + + Future testExtraShutdownsDoNotUnderflow() async { + await expectLater(Veilid.instance.shutdownVeilidCore, + throwsA(isA())); + + final stream = await Veilid.instance.startupVeilidCore(config); + final collector = _UpdateCollector(stream); + await Veilid.instance.shutdownVeilidCore(); + expect(await Veilid.instance.isShutdown(), isTrue); + await collector.done.timeout(_updateTimeout); + + await expectLater(Veilid.instance.shutdownVeilidCore, + throwsA(isA())); + expect(await Veilid.instance.isShutdown(), isTrue); + + // the node comes back after the extra shutdowns + final restarted = await Veilid.instance.startupVeilidCore(config); + final restartedCollector = _UpdateCollector(restarted); + expect(await Veilid.instance.isShutdown(), isFalse); + await Veilid.instance.shutdownVeilidCore(); + await restartedCollector.done.timeout(_updateTimeout); + } + + Future testReapedSinkHoldsNoReference() async { + final firstStream = await Veilid.instance.startupVeilidCore(config); + final secondStream = await Veilid.instance.startupVeilidCore(config); + final first = _UpdateCollector(firstStream); + final second = _UpdateCollector(secondStream); + + await Veilid.instance.attach(); + await second.firstAttachment.timeout(_updateTimeout); + + // cancelling drops the first sink: native reaps the dead port on a later + // post, web deregisters it from onCancel. The parked generator resumes + // only on an update and the mock emits none unprompted, so drive the flow + // with detach/attach transitions instead of waiting on ambient updates. + final cancelled = first.cancel(); + await Veilid.instance.detach(); + await Veilid.instance.attach(); + await cancelled.timeout(_updateTimeout); + + // the cancelled port is closed now, so these posts fail on it and reap it + final delivered = second.nextUpdates(2); + await Veilid.instance.detach(); + await Veilid.instance.attach(); + await delivered.timeout(_updateTimeout); + + // reaping a dead sink is not a shutdown + expect(await Veilid.instance.isShutdown(), isFalse); + await Veilid.instance.detach(); + + // the reaped sink holds no reference, so the survivor is the last one + await Veilid.instance.shutdownVeilidCore(); + expect(await Veilid.instance.isShutdown(), isTrue); + await second.done.timeout(_updateTimeout); + + // the reaped caller's own release finds nothing left to shut down + await expectLater(Veilid.instance.shutdownVeilidCore, + throwsA(isA())); + } +} diff --git a/veilid-flutter/example/lib/app.dart b/veilid-flutter/example/lib/app.dart index b7bc7ea7..a6ff1b3c 100644 --- a/veilid-flutter/example/lib/app.dart +++ b/veilid-flutter/example/lib/app.dart @@ -57,7 +57,8 @@ class _MyAppState extends State with UiLoggy { veilidVersion = 'Failed to get veilid version.'; } - // In case of hot restart shut down first + // A hot restart leaves the running node with no reference held by this + // program, so this is a no-op and the next startup joins that node try { await Veilid.instance.shutdownVeilidCore(); } on Exception { @@ -160,12 +161,15 @@ class _MyAppState extends State with UiLoggy { final updateStream = await Veilid.instance.startupVeilidCore(config); + // joining a running node inherits whatever attachment it already had + final state = await Veilid.instance.getVeilidState(); + setState(() { _updateStream = updateStream; _updateProcessor = processUpdates(); _startedUp = true; _startupChanging = false; - _attached = false; + _attached = state.attachment.state != AttachmentState.detached; _attachChanging = false; }); } else if (!startup && _startedUp) { diff --git a/veilid-flutter/lib/veilid.dart b/veilid-flutter/lib/veilid.dart index 864c3e0d..09c02350 100644 --- a/veilid-flutter/lib/veilid.dart +++ b/veilid-flutter/lib/veilid.dart @@ -86,10 +86,14 @@ abstract class Veilid { /// Start veilid-core with the given [config] and return the stream of /// [VeilidUpdate] events the node emits. /// - /// Blocks on core initialization. The open half of the startup/shutdown pair: - /// must be matched with [shutdownVeilidCore]. Throws if already started. + /// The open half of the startup/shutdown pair: must be matched with + /// [shutdownVeilidCore]. Startup is idempotent and refcounted. When a node + /// started with the same `programName` and `namespace` is already running, + /// this joins it and returns a stream of its own, [config] is ignored, and + /// only core initialization for a genuinely new node blocks. /// - /// Throws [VeilidAPIExceptionAlreadyInitialized] if a node is already running, + /// Throws [VeilidAPIExceptionAlreadyInitialized] if a node with a different + /// `programName` or `namespace` is already running, /// [VeilidAPIExceptionGeneric] if [config] fails validation, /// [VeilidAPIExceptionInternal] if a subsystem fails to initialize. Fatal for /// this call; fix the config or environment and retry. @@ -108,24 +112,29 @@ abstract class Veilid { /// Connect to the network. /// /// Sets the maintain-peers flag; the network connect proceeds in the - /// background. Throws if already attached. + /// background. Idempotent: a node several callers share is attached by all of + /// them, so finding it already attached succeeds. /// - /// Throws [VeilidAPIExceptionGeneric] if already attached, or - /// [VeilidAPIExceptionNotInitialized] if Veilid has been shut down. + /// Throws [VeilidAPIExceptionNotInitialized] if Veilid has been shut down. Future attach(); /// Disconnect from the network. /// /// Clears the maintain-peers flag; the network teardown proceeds in the - /// background. Throws if already detached. + /// background. Idempotent, like [attach]: finding the node already detached + /// succeeds. /// - /// Throws [VeilidAPIExceptionGeneric] if already detached, or - /// [VeilidAPIExceptionNotInitialized] if Veilid has been shut down. + /// Throws [VeilidAPIExceptionNotInitialized] if Veilid has been shut down. Future detach(); - /// Shut down Veilid and terminate the API. + /// Release this caller's reference to the running node, shutting it down and + /// terminating the API when the last reference goes. /// - /// Blocks until the core has finished shutting down. + /// Releases one reference per [startupVeilidCore], most recent first, and + /// closes that call's update stream. Only the release of the last reference + /// shuts the node down and blocks until the core has finished doing so; while + /// another reference is outstanding this returns with the node still running. + /// A caller holding no reference succeeds without doing anything. /// /// Throws [VeilidAPIExceptionNotInitialized] if Veilid was never started or /// has already been shut down. diff --git a/veilid-flutter/lib/veilid_ffi.dart b/veilid-flutter/lib/veilid_ffi.dart index 36ff1586..ae613ed4 100644 --- a/veilid-flutter/lib/veilid_ffi.dart +++ b/veilid-flutter/lib/veilid_ffi.dart @@ -61,7 +61,7 @@ void _eagerResolveNativeSymbols() { Native.addressOf>(VeilidFFI._getVeilidState), Native.addressOf>(VeilidFFI._attach), Native.addressOf>(VeilidFFI._detach), - Native.addressOf>(VeilidFFI._shutdownVeilidCore), + Native.addressOf>(VeilidFFI._shutdownVeilidCore), Native.addressOf>(VeilidFFI._routingContext), Native.addressOf>(VeilidFFI._releaseRoutingContext), Native.addressOf>(VeilidFFI._routingContextWithDefaultSafety), @@ -409,7 +409,12 @@ Stream processStreamJson( ) async* { try { await for (final value in port) { - final list = value as List; + // rust probes a suspect port with a bare int to tell a dead port from a + // payload the dart serializer refused + if (value is! List) { + continue; + } + final list = value; switch (list[0] as int) { case messageStreamItemJson: { @@ -1706,6 +1711,11 @@ class VeilidCryptoSystemFFI extends VeilidCryptoSystem { // FFI implementation of high level Veilid API class VeilidFFI extends Veilid { + + // One stream port per reference this isolate holds on the running node. + // Shutdown releases the most recent one. + final _updateStreamPorts = []; + // Shared library functions @Native)>(symbol: 'free_string') external static void _freeString(Pointer s); @@ -1750,8 +1760,8 @@ class VeilidFFI extends Veilid { @Native(symbol: 'detach') external static void _detach(int port); - @Native(symbol: 'shutdown_veilid_core') - external static void _shutdownVeilidCore(int port); + @Native(symbol: 'shutdown_veilid_core') + external static void _shutdownVeilidCore(int port, int streamPort); @Native(symbol: 'routing_context') external static void _routingContext(int port); @@ -2544,7 +2554,7 @@ class VeilidFFI extends Veilid { } @override - Future> startupVeilidCore(VeilidConfig config) { + Future> startupVeilidCore(VeilidConfig config) async { final nativeConfig = jsonEncode(config).toNativeUtf8(); final recvStreamPort = ReceivePort('veilid_api_stream'); final sendStreamPort = recvStreamPort.sendPort; @@ -2556,10 +2566,19 @@ class VeilidFFI extends Veilid { nativeConfig, ); malloc.free(nativeConfig); - return processFutureStream( - processStreamJson(VeilidUpdate.fromJson, recvStreamPort), - recvPort.first, - ); + try { + final updateStream = await processFutureStream( + processStreamJson(VeilidUpdate.fromJson, recvStreamPort), + recvPort.first, + ); + _updateStreamPorts.add(sendStreamPort.nativePort); + return updateStream; + } on Exception { + // processStreamJson is lazy: nobody subscribes to a failed startup, so + // its cleanup never runs and the port would pin the event loop + recvStreamPort.close(); + rethrow; + } } @override @@ -2598,7 +2617,12 @@ class VeilidFFI extends Veilid { Future shutdownVeilidCore() { final recvPort = ReceivePort('shutdown_veilid_core'); final sendPort = recvPort.sendPort; - _shutdownVeilidCore(sendPort.nativePort); + // the reply port is one-shot and carries no identity, so the stream port is + // what tells rust whose reference to release + final streamPort = _updateStreamPorts.isEmpty + ? 0 + : _updateStreamPorts.removeLast(); + _shutdownVeilidCore(sendPort.nativePort, streamPort); return processFutureVoid(recvPort.first); } diff --git a/veilid-flutter/lib/veilid_js.dart b/veilid-flutter/lib/veilid_js.dart index 14703cb4..969e2ec0 100644 --- a/veilid-flutter/lib/veilid_js.dart +++ b/veilid-flutter/lib/veilid_js.dart @@ -40,6 +40,9 @@ String _wasmCallString(String method, [List? arguments]) => int _wasmCallInt(String method, [List? arguments]) => wasm.callMethodVarArgs(method.toJS, arguments).toDartInt; +bool _wasmCallBool(String method, [List? arguments]) => + wasm.callMethodVarArgs(method.toJS, arguments).toDart; + void _wasmCallVoid(String method, [List? arguments]) => wasm.callMethodVarArgs(method.toJS, arguments); @@ -993,6 +996,46 @@ class VeilidTableDBJS extends VeilidTableDB { // JS implementation of high level Veilid API class VeilidJS extends Veilid { + // Names this dart program to wasm. A hot restart resets lazy statics while the + // wasm module and its node survive, so the next program gets a new value and + // wasm can retire the update sinks the old one left registered. + static final String _generation = + DateTime.now().microsecondsSinceEpoch.toString(); + + // One caller id per reference this dart context holds on the running node. + // Shutdown releases the most recent one. Wasm owns the refcount and every + // lifecycle decision. + final _callerIds = []; + + static int _callerIdSequence = 0; + + static String _newCallerId() => '$_generation-${_callerIdSequence++}'; + + // Wasm delivers a serialized update, or null to close the stream. Throwing out + // of here is wasm's only evidence that the js context is gone, so a decode + // failure against a newer wasm blob must not escape. + static JSFunction _updateSinkCallback(StreamController sink) => + ((JSString? update) { + if (sink.isClosed) { + return; + } + if (update == null) { + // never awaited: close() on a controller nobody listened to completes + // only once a listener arrives to take the done event + unawaited(sink.close()); + return; + } + try { + sink.add( + VeilidUpdate.fromJson( + jsonDecode(update.toDart) as Map, + ), + ); + } on Object catch (e) { + web.console.error('veilid: dropped an undecodable update: $e'.toJS); + } + }).toJS; + @override void initializeVeilidCore(Map platformConfigJson) { final platformConfigJsonString = jsonEncode(platformConfigJson); @@ -1018,31 +1061,42 @@ class VeilidJS extends Veilid { @override Future> startupVeilidCore(VeilidConfig config) async { - final streamController = StreamController(); - void updateCallback(String update) { - final updateJson = jsonDecode(update) as Map; - if (updateJson['kind'] == 'Shutdown') { - unawaited(streamController.close()); - } else { - final update = VeilidUpdate.fromJson(updateJson); - streamController.add(update); - } + final configJson = jsonEncode(config); + final callerId = _newCallerId(); + + // an abandoned subscription must not pin the refcount + late final StreamController sink; + sink = StreamController( + onCancel: () { + _callerIds.remove(callerId); + _wasmCallVoid('deregister_update_sink', [callerId.toJS]); + }, + ); + + try { + await _wasmPromiseVoid('startup_veilid_core', [ + callerId.toJS, + _generation.toJS, + _updateSinkCallback(sink), + configJson.toJS, + ]); + } on Exception { + unawaited(sink.close()); + rethrow; } + _callerIds.add(callerId); - await _wasmPromiseVoid('startup_veilid_core', [ - updateCallback.toJS, - jsonEncode(config).toJS, - ]); - - return streamController.stream; + return sink.stream; } @override Future getVeilidState() => _wasmPromiseJson('get_veilid_state', VeilidState.fromJson, []); + // is_shutdown is a plain wasm export, not a promise like the rest of the + // lifecycle calls @override - Future isShutdown() => _wasmPromiseBool('is_shutdown', []); + Future isShutdown() async => _wasmCallBool('is_shutdown', []); @override Future attach() => _wasmPromiseVoid('attach', []); @@ -1051,8 +1105,16 @@ class VeilidJS extends Veilid { Future detach() => _wasmPromiseVoid('detach', []); @override - Future shutdownVeilidCore() => - _wasmPromiseVoid('shutdown_veilid_core', []); + Future shutdownVeilidCore() { + // an empty id matches no registration, so wasm reports NotInitialized for a + // node that was never started and no-ops for a caller holding no reference, + // in both cases after retiring what an earlier dart program left registered + final callerId = _callerIds.isEmpty ? '' : _callerIds.removeLast(); + return _wasmPromiseVoid('shutdown_veilid_core', [ + callerId.toJS, + _generation.toJS, + ]); + } @override List validCryptoKinds() => _wasmCallJson( @@ -1249,8 +1311,11 @@ class VeilidJS extends Veilid { } @override - List veilidFeatures() => - _wasmCallJson('veilid_features', (x) => x as List, []); + List veilidFeatures() => _wasmCallJson( + 'veilid_features', + (x) => (x as List).map((e) => e as String).toList(), + [], + ); @override String defaultVeilidConfig() => _wasmCallString('default_veilid_config', []); diff --git a/veilid-flutter/packages/veilid_test/lib/src/veilid_fixture.dart b/veilid-flutter/packages/veilid_test/lib/src/veilid_fixture.dart index 2cce2cb9..bc1dfcb5 100644 --- a/veilid-flutter/packages/veilid_test/lib/src/veilid_fixture.dart +++ b/veilid-flutter/packages/veilid_test/lib/src/veilid_fixture.dart @@ -293,13 +293,16 @@ class DefaultVeilidFixture implements VeilidFixture { Future tearDown() async { assert(_fixtureMutex.isLocked, 'should not tearDown without setUp'); - await Veilid.instance.shutdownVeilidCore(); - await _veilidUpdateSubscription?.cancel(); - await _updateStreamController.close(); + try { + await Veilid.instance.shutdownVeilidCore(); + await _veilidUpdateSubscription?.cancel(); + await _updateStreamController.close(); - _veilidUpdateSubscription = null; - _veilidUpdateStream = null; - - _fixtureMutex.release(); + _veilidUpdateSubscription = null; + _veilidUpdateStream = null; + } finally { + // an unreleased static mutex turns one failed test into a hung run + _fixtureMutex.release(); + } } } diff --git a/veilid-flutter/rust/Cargo.toml b/veilid-flutter/rust/Cargo.toml index 41de4725..1eeb6664 100644 --- a/veilid-flutter/rust/Cargo.toml +++ b/veilid-flutter/rust/Cargo.toml @@ -77,7 +77,8 @@ opentelemetry-semantic-conventions = { version = "0.31.0", optional = true } async-std = { version = "1.13.2", features = ["unstable"], optional = true } tokio = { version = "1.50.0", optional = true } -allo-isolate = "0.1.27" +# zero-copy off: a failed post must not run rust finalizers inline on the posting thread +allo-isolate = { version = "0.1.27", default-features = false } ffi-support = "0.4.4" lazy_static = "1.5.0" hostname = "0.4.2" diff --git a/veilid-flutter/rust/src/dart_ffi.rs b/veilid-flutter/rust/src/dart_ffi.rs index 927dbfd3..d4e9d08f 100644 --- a/veilid-flutter/rust/src/dart_ffi.rs +++ b/veilid-flutter/rust/src/dart_ffi.rs @@ -121,6 +121,7 @@ lazy_static! { static ref DHT_TRANSACTIONS: Mutex> = Mutex::new(BTreeMap::new()); static ref FLAME_GUARD: Mutex>>> = Mutex::new(None); + static ref NODE_STATE: Mutex> = Mutex::new(None); } async fn get_veilid_api() -> VeilidAPIResult { @@ -131,6 +132,54 @@ async fn get_veilid_api() -> VeilidAPIResult { .ok_or(VeilidAPIError::NotInitialized) } +/// veilid-core's per-node identity, `(program_name, namespace)`. +type InitKey = (String, String); + +/// Shim state for the running node: who receives its updates, and what started it. +/// +/// The number of registered sinks is the node's reference count. Startup registers one and +/// shutdown retires one, and the node is torn down when an explicit shutdown retires the last. +/// Reaping a dead sink is not a shutdown: an Android foreground service outliving its activity's +/// engine must keep the node running with nothing attached. +/// +/// Every transition runs under the `VEILID_API` async lock. Racing two `shutdown().await` calls +/// would let the loser report completion before the node is down, and core's own Arc count is not +/// a substitute, since `get_veilid_api` clones it on every FFI call. +/// +/// One instance today, shaped to become one per `InitKey`. +struct NodeState { + sinks: Arc, + config_json: String, + init_key: InitKey, +} + +/// The fields of a startup config that name which node it starts. +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct VeilidFFIConfigInitKey { + program_name: String, + namespace: String, +} + +/// Read the init key from a startup config without consuming the config, so core still gets it +/// verbatim and still warns about unknown keys. +fn peek_init_key(config_json: &str) -> VeilidAPIResult { + let key = veilid_core::deserialize_json::(config_json)?; + if key.program_name.is_empty() { + return Err(VeilidAPIError::generic( + "Program name must not be empty in 'program_name'", + )); + } + Ok((key.program_name, key.namespace)) +} + +fn update_callback(sinks: Arc) -> UpdateCallback { + Arc::new(move |update: VeilidUpdate| match update { + VeilidUpdate::Shutdown => sinks.close_all(), + _ => sinks.broadcast_json(update), + }) +} + ///////////////////////////////////////// // FFI Helpers @@ -544,7 +593,6 @@ pub extern "C" fn change_log_ignore(layer: FfiStr, log_ignore: FfiStr) { #[instrument(level = "trace", target = "ffi", skip_all)] pub extern "C" fn startup_veilid_core(port: i64, stream_port: i64, config: FfiStr) { let config = config.into_opt_string(); - let stream = DartIsolateStream::new(stream_port); DartIsolateWrapper::new(port).spawn_result( async move { let config_json = match config { @@ -557,29 +605,48 @@ pub extern "C" fn startup_veilid_core(port: i64, stream_port: i64, config: FfiSt return VeilidAPIResult::Err(err); } }; + let init_key = peek_init_key(&config_json)?; let mut api_lock = VEILID_API.lock().await; + + // join the running node when it is the same one this config asks for if api_lock.is_some() { - return VeilidAPIResult::Err(VeilidAPIError::AlreadyInitialized); + let mut node_state = NODE_STATE.lock(); + let Some(node_state) = node_state.as_mut() else { + return VeilidAPIResult::Err(VeilidAPIError::NotInitialized); + }; + if node_state.init_key != init_key { + return VeilidAPIResult::Err(VeilidAPIError::AlreadyInitialized); + } + if node_state.config_json != config_json { + warn!( + target: "ffi", + "joining a node started with a different config, this config is ignored" + ); + } + node_state.sinks.register(stream_port); + return VeilidAPIResult::Ok(()); } - let sink = stream.clone(); - let update_callback = Arc::new(move |update: VeilidUpdate| { - let sink = sink.clone(); - match update { - VeilidUpdate::Shutdown => { - sink.close(); + let sinks = Arc::new(DartUpdateSinkSet::new()); + sinks.register(stream_port); + + let veilid_api = + match api_startup_json(update_callback(sinks.clone()), config_json.clone()).await { + Ok(v) => v, + Err(e) => { + sinks.retire(stream_port); + return VeilidAPIResult::Err(e); } - _ => { - sink.item_json(update); - } - } + }; + *api_lock = Some(veilid_api); + *NODE_STATE.lock() = Some(NodeState { + sinks, + config_json, + init_key, }); - let veilid_api = api_startup_json(update_callback, config_json).await?; - *api_lock = Some(veilid_api); - - Ok(()) + VeilidAPIResult::Ok(()) } .in_current_span(), ); @@ -621,7 +688,13 @@ pub extern "C" fn attach(port: i64) { DartIsolateWrapper::new(port).spawn_result( async move { let veilid_api = get_veilid_api().await?; - veilid_api.attach().await?; + // every caller sharing the node attaches it, so core's "already attached" Generic + // (its only error here other than NotInitialized) is this caller's success + if let Err(e) = veilid_api.attach().await { + if !matches!(e, VeilidAPIError::Generic { .. }) { + return VeilidAPIResult::Err(e); + } + } VeilidAPIResult::Ok(()) } .in_current_span(), @@ -634,7 +707,12 @@ pub extern "C" fn detach(port: i64) { DartIsolateWrapper::new(port).spawn_result( async move { let veilid_api = get_veilid_api().await?; - veilid_api.detach().await?; + // symmetric with attach: "already detached" is another caller's doing, not an error + if let Err(e) = veilid_api.detach().await { + if !matches!(e, VeilidAPIError::Generic { .. }) { + return VeilidAPIResult::Err(e); + } + } VeilidAPIResult::Ok(()) } .in_current_span(), @@ -643,19 +721,56 @@ pub extern "C" fn detach(port: i64) { #[no_mangle] #[instrument(level = "trace", target = "ffi", skip_all)] -pub extern "C" fn shutdown_veilid_core(port: i64) { +pub extern "C" fn shutdown_veilid_core(port: i64, stream_port: i64) { DartIsolateWrapper::new(port).spawn_result( async move { let mut api_lock = VEILID_API.lock().await; - let veilid_api = api_lock.take().ok_or(VeilidAPIError::NotInitialized)?; - // A dead Dart session (hot restart) never releases its handles; - // drop them here so the table store can close its files instead - // of leaving sqlite pools pinned open into the next startup + if api_lock.is_none() { + return VeilidAPIResult::Err(VeilidAPIError::NotInitialized); + } + + let sinks = { + let node_state = NODE_STATE.lock(); + let Some(node_state) = node_state.as_ref() else { + return VeilidAPIResult::Err(VeilidAPIError::NotInitialized); + }; + // a caller with no registered sink holds no reference + if !node_state.sinks.contains(stream_port) { + return VeilidAPIResult::Ok(()); + } + node_state.sinks.clone() + }; + + // a dead engine's sink counts until something reaps it, and a detached node can go + // arbitrarily long without an update to reap it on + sinks.reap_dead_except(stream_port); + + if sinks.len() > 1 { + sinks.retire(stream_port); + return VeilidAPIResult::Ok(()); + } + + // the last reference stays registered through termination so its consumer sees the + // updates core emits on the way down + let Some(veilid_api) = api_lock.take() else { + return VeilidAPIResult::Err(VeilidAPIError::NotInitialized); + }; + // hot restart never releases handles; drop them so sqlite pools do not stay pinned + // into the next startup. Last reference only - a live caller's handles are still its own TABLE_DB_TRANSACTIONS.lock().clear(); TABLE_DBS.lock().clear(); DHT_TRANSACTIONS.lock().clear(); ROUTING_CONTEXTS.lock().clear(); veilid_api.shutdown().await; + + // unconditional: a teardown that emits no Shutdown update would park every consumer + // in `await for` forever + sinks.close_all(); + // dropped out of the guard's scope: retiring a sink posts, and posting under a lock + // is how this deadlocks + let retired_state = NODE_STATE.lock().take(); + drop(retired_state); + VeilidAPIResult::Ok(()) } .in_current_span(), diff --git a/veilid-flutter/rust/src/dart_isolate_wrapper.rs b/veilid-flutter/rust/src/dart_isolate_wrapper.rs index e722e68a..ef9c573e 100644 --- a/veilid-flutter/rust/src/dart_isolate_wrapper.rs +++ b/veilid-flutter/rust/src/dart_isolate_wrapper.rs @@ -1,3 +1,18 @@ +//! Posting to Dart from Rust. +//! +//! [`DartIsolateWrapper`] answers one FFI call on a one-shot port. [`DartUpdateSinkSet`] fans the +//! veilid-core update callback out to every Dart update stream registered against the running +//! node. Rules the sink set rests on: +//! +//! - Never post while holding the sink lock. Disposing a Dart stream posts a close message, and a +//! synchronous dispose racing an in-flight post deadlocks. Take what you need out under the +//! lock, release, then post. +//! - Posting from a tokio thread after every Flutter engine has died is defined only because +//! Flutter leaks a strong VM reference (`leak_vm = true`) so `Dart_Cleanup` never runs; +//! `Dart_PostCObject` is contractually legal only before `Dart_Cleanup`. +//! - A Dart consumer parked in `await for` wakes only on a close message, so every path that +//! retires a sink posts one. + use crate::tools::*; pub use allo_isolate::IntoDart; @@ -6,7 +21,7 @@ use core::fmt::Debug; use core::future::Future; use parking_lot::Mutex; use serde::*; -use std::sync::Arc; +use std::collections::BTreeMap; #[derive(Clone)] pub struct DartIsolateWrapper { @@ -94,80 +109,193 @@ impl DartIsolateWrapper { } } -struct DartIsolateStreamInner { - pub isolate: Option, +/// One registered Dart update stream, identified by the ReceivePort it posts to. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct DartUpdateSink { + port: i64, } -impl Drop for DartIsolateStreamInner { - fn drop(&mut self) { - if let Some(isolate) = self.isolate { - isolate.post(vec![MESSAGE_STREAM_CLOSE.into_dart()]); - } - } -} - -#[derive(Clone)] -pub struct DartIsolateStream { - inner: Arc>, -} - -impl DartIsolateStream { +impl DartUpdateSink { pub fn new(port: i64) -> Self { - DartIsolateStream { - inner: Arc::new(Mutex::new(DartIsolateStreamInner { - isolate: Some(Isolate::new(port)), - })), - } + DartUpdateSink { port } } - // pub fn item(&self, value: T) -> bool { - // let mut inner = self.inner.lock(); - // if let Some(isolate) = inner.isolate.take() { - // isolate.post(vec![MESSAGE_STREAM_ITEM.into_dart(), value.into_dart()]) - // } else { - // false - // } - // } - - pub fn item_json(&self, value: T) -> bool { - let inner = self.inner.lock(); - if let Some(isolate) = &inner.isolate { - isolate.post(vec![ - MESSAGE_STREAM_ITEM_JSON.into_dart(), - veilid_core::serialize_json(value).into_dart(), - ]) - } else { - false - } + pub fn port(&self) -> i64 { + self.port } - // pub fn abort(self, error: E) -> bool { - // let mut inner = self.inner.lock(); - // if let Some(isolate) = inner.isolate.take() { - // isolate.post(vec![MESSAGE_STREAM_ABORT.into_dart(), error.into_dart()]) - // } else { - // false - // } - // } + pub fn item_json(&self, json: String) -> bool { + Isolate::new(self.port).post(vec![MESSAGE_STREAM_ITEM_JSON.into_dart(), json.into_dart()]) + } - // pub fn abort_json(self, error: E) -> bool { - // let mut inner = self.inner.lock(); - // if let Some(isolate) = inner.isolate.take() { - // isolate.post(vec![ - // MESSAGE_STREAM_ABORT_JSON.into_dart(), - // veilid_core::serialize_json(error).into_dart(), - // ]) - // } else { - // false - // } - // } + pub fn close(&self) -> bool { + Isolate::new(self.port).post(vec![MESSAGE_STREAM_CLOSE.into_dart()]) + } +} - pub fn close(self) -> bool { - let mut inner = self.inner.lock(); - if let Some(isolate) = inner.isolate.take() { - isolate.post(vec![MESSAGE_STREAM_CLOSE.into_dart()]) - } else { - false +/// The set of Dart update streams a running node fans its updates out to. +/// +/// Shared between the FFI shim and the `UpdateCallback` handed to veilid-core, which posts through +/// it from tokio threads for the life of the node. +#[derive(Default)] +pub struct DartUpdateSinkSet { + sinks: Mutex>, +} + +impl Drop for DartUpdateSinkSet { + fn drop(&mut self) { + for sink in self.sinks.get_mut().values() { + sink.close(); } } } + +impl DartUpdateSinkSet { + pub fn new() -> Self { + Self::default() + } + + /// Register a sink for `port`, returning whether it was not already registered. + pub fn register(&self, port: i64) -> bool { + self.sinks + .lock() + .insert(port, DartUpdateSink::new(port)) + .is_none() + } + + /// Deregister `port` and close its stream, returning whether it was registered. + pub fn retire(&self, port: i64) -> bool { + let sink = self.sinks.lock().remove(&port); + let Some(sink) = sink else { + return false; + }; + sink.close(); + true + } + + pub fn contains(&self, port: i64) -> bool { + self.sinks.lock().contains_key(&port) + } + + pub fn len(&self) -> usize { + self.sinks.lock().len() + } + + /// Post `value` to every registered sink, reaping the ones whose port is proven gone. + pub fn broadcast_json(&self, value: T) { + let sinks: Vec = self.sinks.lock().values().copied().collect(); + if sinks.is_empty() { + return; + } + let json = veilid_core::serialize_json(value); + let mut dead = Vec::::new(); + for sink in sinks { + // a failed update post is not proof the port is gone, see dart_port_is_live + if !sink.item_json(json.clone()) && !dart_port_is_live(sink.port()) { + dead.push(sink.port()); + } + } + if dead.is_empty() { + return; + } + let mut sinks = self.sinks.lock(); + for port in dead { + sinks.remove(&port); + } + } + + /// Drop every registered sink whose port is proven gone, leaving `keep` alone. + /// + /// A caller releasing its own reference has not proven its port is live, so `keep` is never + /// reaped out from under it. + pub fn reap_dead_except(&self, keep: i64) { + let ports: Vec = self + .sinks + .lock() + .keys() + .copied() + .filter(|port| *port != keep) + .collect(); + let dead: Vec = ports + .into_iter() + .filter(|port| !dart_port_is_live(*port)) + .collect(); + if dead.is_empty() { + return; + } + let mut sinks = self.sinks.lock(); + for port in dead { + sinks.remove(&port); + } + } + + /// Deregister every sink and close its stream. + pub fn close_all(&self) { + let sinks = core::mem::take(&mut *self.sinks.lock()); + for sink in sinks.values() { + sink.close(); + } + } +} + +/// A Dart port that has left the VM port map rejects every post. +/// +/// The int32 arm of Dart's message serializer has no failure path, so a rejected primitive proves +/// the port is gone. A rejected update does not: Dart can fail to serialize a String for encoding +/// reasons (invalid utf8, invalid string length) on a perfectly live port. +pub fn dart_port_is_live(port: i64) -> bool { + port != 0 && Isolate::new(port).post(0i32) +} + +#[cfg(test)] +mod tests { + use super::*; + + // With no Dart VM attached every post fails, which is the dead-port path. + const PORT: i64 = 42; + + #[test] + fn register_and_retire() { + let sinks = DartUpdateSinkSet::new(); + assert_eq!(sinks.len(), 0); + assert!(!sinks.retire(PORT)); + + assert!(sinks.register(PORT)); + assert!(!sinks.register(PORT)); + assert!(sinks.register(PORT + 1)); + assert_eq!(sinks.len(), 2); + assert!(sinks.contains(PORT)); + + assert!(sinks.retire(PORT)); + assert!(!sinks.retire(PORT)); + assert_eq!(sinks.len(), 1); + + sinks.close_all(); + assert_eq!(sinks.len(), 0); + } + + #[test] + fn reap_leaves_the_kept_port() { + let sinks = DartUpdateSinkSet::new(); + sinks.register(PORT); + sinks.register(PORT + 1); + sinks.reap_dead_except(PORT); + assert_eq!(sinks.len(), 1); + assert!(sinks.contains(PORT)); + } + + #[test] + fn broadcast_reaps_dead_ports() { + let sinks = DartUpdateSinkSet::new(); + sinks.register(PORT); + sinks.register(PORT + 1); + sinks.broadcast_json("update"); + assert_eq!(sinks.len(), 0); + } + + #[test] + fn zero_port_is_never_live() { + assert!(!dart_port_is_live(0)); + assert!(!dart_port_is_live(PORT)); + } +} diff --git a/veilid-flutter/rust/src/dart_js_wrapper.rs b/veilid-flutter/rust/src/dart_js_wrapper.rs new file mode 100644 index 00000000..bc3ed70f --- /dev/null +++ b/veilid-flutter/rust/src/dart_js_wrapper.rs @@ -0,0 +1,173 @@ +//! Posting to Dart from Rust, wasm flavor. +//! +//! [`DartUpdateSinkSet`] fans the veilid-core update callback out to every Dart update stream +//! registered against the running node, the sibling of the FFI shim's sink set. Rules it rests on: +//! +//! - Never call into Dart while holding the sink borrow. A Dart handler reenters wasm (a cancelled +//! subscription deregisters itself), and a live `RefCell` borrow turns that into a panic. Copy +//! out, call, then mutate. +//! - A Dart consumer parked in `await for` wakes only on a close, so every path that drops a sink +//! calls its callback with null. +//! - There is no web analogue of the FFI port liveness probe. A Dart hot restart leaves the +//! previous program's interop closures callable, so a leftover sink keeps accepting updates and +//! cannot be detected by calling it. Staleness is decided by the generation id of the Dart +//! program that registered the sink, resting on one Dart program per page. A throwing callback +//! is only the backstop for a JS context that is genuinely gone. + +use super::*; + +/// One registered Dart update stream, keyed in the set by caller id and tagged with the +/// generation of the Dart program that registered it. +#[derive(Clone)] +pub struct DartUpdateSink { + generation: String, + callback: Function, +} + +impl DartUpdateSink { + pub fn new(generation: String, callback: Function) -> Self { + DartUpdateSink { + generation, + callback, + } + } + + /// Deliver one serialized update, returning whether the Dart handler accepted it. + pub fn item_json(&self, json: &str) -> bool { + self.callback + .call1(&JsValue::UNDEFINED, &JsValue::from_str(json)) + .is_ok() + } + + /// Close the Dart stream. A null update is the close message. + pub fn close(&self) -> bool { + self.callback + .call1(&JsValue::UNDEFINED, &JsValue::NULL) + .is_ok() + } +} + +/// The set of Dart update streams a running node fans its updates out to. +/// +/// Shared between the wasm shim and the `UpdateCallback` handed to veilid-core. The `SendWrapper` +/// is what lets a single-threaded `RefCell` satisfy the callback's `Send + Sync` bound. +pub struct DartUpdateSinkSet { + sinks: SendWrapper>>, +} + +impl Default for DartUpdateSinkSet { + fn default() -> Self { + DartUpdateSinkSet { + sinks: SendWrapper::new(RefCell::new(BTreeMap::new())), + } + } +} + +impl Drop for DartUpdateSinkSet { + fn drop(&mut self) { + for sink in self.sinks.get_mut().values() { + sink.close(); + } + } +} + +impl DartUpdateSinkSet { + pub fn new() -> Self { + Self::default() + } + + /// Register a sink for `caller_id`, returning whether it was not already registered. + pub fn register(&self, caller_id: String, generation: String, callback: Function) -> bool { + self.sinks + .borrow_mut() + .insert(caller_id, DartUpdateSink::new(generation, callback)) + .is_none() + } + + /// Retire every sink a Dart program other than `generation` registered. + /// + /// A hot restart replaces the Dart program while this module and its node carry on, so those + /// sinks belong to a context that no longer exists. Closing them ends the streams the old + /// context is still consuming. + pub fn retire_other_generations(&self, generation: &str) { + let stale: Vec = { + let mut sinks = self.sinks.borrow_mut(); + let stale_ids: Vec = sinks + .iter() + .filter(|(_, sink)| sink.generation != generation) + .map(|(caller_id, _)| caller_id.clone()) + .collect(); + stale_ids + .iter() + .filter_map(|caller_id| sinks.remove(caller_id)) + .collect() + }; + for sink in stale { + sink.close(); + } + } + + /// Deregister `caller_id` and close its stream, returning whether it was registered. + pub fn retire(&self, caller_id: &str) -> bool { + let sink = self.sinks.borrow_mut().remove(caller_id); + let Some(sink) = sink else { + return false; + }; + sink.close(); + true + } + + /// Deregister `caller_id` without closing its stream, for a consumer that already ended it. + pub fn deregister(&self, caller_id: &str) -> bool { + self.sinks.borrow_mut().remove(caller_id).is_some() + } + + pub fn contains(&self, caller_id: &str) -> bool { + self.sinks.borrow().contains_key(caller_id) + } + + pub fn len(&self) -> usize { + self.sinks.borrow().len() + } + + /// Post `value` to every registered sink, retiring the ones whose callback throws. + pub fn broadcast_json(&self, value: T) { + let sinks: Vec<(String, DartUpdateSink)> = self + .sinks + .borrow() + .iter() + .map(|(caller_id, sink)| (caller_id.clone(), sink.clone())) + .collect(); + if sinks.is_empty() { + return; + } + let json = veilid_core::serialize_json(value); + let mut dead = Vec::::new(); + for (caller_id, sink) in sinks { + // the dart handler swallows its own decode failures, so a throw is a broken js context + if !sink.item_json(&json) { + dead.push(caller_id); + } + } + if dead.is_empty() { + return; + } + let dead: Vec = { + let mut sinks = self.sinks.borrow_mut(); + dead.iter() + .filter_map(|caller_id| sinks.remove(caller_id)) + .collect() + }; + for sink in dead { + sink.close(); + } + } + + /// Deregister every sink and close its stream. + pub fn close_all(&self) { + let sinks = core::mem::take(&mut *self.sinks.borrow_mut()); + for sink in sinks.values() { + sink.close(); + } + } +} diff --git a/veilid-flutter/rust/src/dart_wasm.rs b/veilid-flutter/rust/src/dart_wasm.rs index 3c286bb6..9a9422f5 100644 --- a/veilid-flutter/rust/src/dart_wasm.rs +++ b/veilid-flutter/rust/src/dart_wasm.rs @@ -1,6 +1,7 @@ /// Veilid WASM Bindings for Flutter/Dart /// The Flutter/Dart bindings are in this lib.rs directly use super::*; +use tracing::warn; #[wasm_bindgen] extern "C" { @@ -27,6 +28,9 @@ lazy_static! { SendWrapper::new(RefCell::new(BTreeMap::new())); static ref TABLE_DB_TRANSACTIONS: SendWrapper>> = SendWrapper::new(RefCell::new(BTreeMap::new())); + static ref NODE_STATE: SendWrapper>> = + SendWrapper::new(RefCell::new(None)); + static ref LIFECYCLE_LOCK: AsyncMutex<()> = AsyncMutex::new(()); } fn get_veilid_api() -> Result { @@ -42,6 +46,56 @@ fn take_veilid_api() -> Result, + config_json: String, +} + +/// The fields of a startup config that name which node it starts. +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct VeilidWASMConfigInitKey { + program_name: String, + namespace: String, +} + +/// Read the init key from a startup config without consuming the config, so core still gets it +/// verbatim and still warns about unknown keys. +fn peek_init_key(config_json: &str) -> VeilidAPIResult { + let key = veilid_core::deserialize_json::(config_json)?; + if key.program_name.is_empty() { + return Err(veilid_core::VeilidAPIError::generic( + "Program name must not be empty in 'program_name'", + )); + } + Ok((key.program_name, key.namespace)) +} + +/// The init key of the node that is actually running. +fn running_init_key(veilid_api: &veilid_core::VeilidAPI) -> VeilidAPIResult { + let config = veilid_api.config()?; + Ok((config.program_name.clone(), config.namespace.clone())) +} + +fn update_callback(sinks: Arc) -> UpdateCallback { + Arc::new(move |update: VeilidUpdate| match update { + VeilidUpdate::Shutdown => sinks.close_all(), + _ => sinks.broadcast_json(update), + }) +} + // Serialize a value to JSON and wrap it as a JS string value. fn to_json(val: T) -> JsValue { JsValue::from_str(&serialize_json(val)) @@ -209,31 +263,88 @@ pub fn change_log_ignore(layer: String, log_ignore: String) { } } +/// Start or join the node this config names, registering `caller_id`'s update sink on it. +/// +/// `caller_id` identifies one Dart consumer for the life of its update stream, and is what +/// `shutdown_veilid_core` releases. `generation` identifies the Dart program that minted it, so a +/// sink left behind by a hot-restarted program can be told apart from a live one. The Dart side +/// mints both; wasm only compares them. #[wasm_bindgen()] -pub fn startup_veilid_core(update_callback_js: Function, json_config: String) -> Promise { - let update_callback_js = SendWrapper::new(update_callback_js); +pub fn startup_veilid_core( + caller_id: String, + generation: String, + update_callback_js: Function, + json_config: String, +) -> Promise { wrap_api_future_void(async move { - let update_callback = Arc::new(move |update: VeilidUpdate| { - let _ret = - match Function::call1(&update_callback_js, &JsValue::UNDEFINED, &to_json(update)) { - Ok(v) => v, - Err(e) => { - console_log(&format!("calling update callback failed: {:?}", e)); - return; - } - }; - }); + let init_key = peek_init_key(&json_config)?; - if VEILID_API.borrow().is_some() { - return Err(veilid_core::VeilidAPIError::AlreadyInitialized); + // held across the startup await, the FFI shim's api lock does the same: two callers racing + // a fresh start would otherwise both reach api_startup_json and the loser would take core's + // AlreadyInitialized for a config that should have joined + let _lifecycle = LIFECYCLE_LOCK.lock().await; + + // join the running node when it is the same one this config asks for + if let Ok(veilid_api) = get_veilid_api() { + if running_init_key(&veilid_api)? != init_key { + return Err(veilid_core::VeilidAPIError::AlreadyInitialized); + } + let sinks = { + let node_state = NODE_STATE.borrow(); + let Some(node_state) = node_state.as_ref() else { + return Err(veilid_core::VeilidAPIError::NotInitialized); + }; + if node_state.config_json != json_config { + warn!( + target: "wasm", + "joining a node started with a different config, this config is ignored" + ); + } + node_state.sinks.clone() + }; + sinks.retire_other_generations(&generation); + sinks.register(caller_id, generation, update_callback_js); + return VeilidAPIResult::Ok(()); } - let veilid_api = veilid_core::api_startup_json(update_callback, json_config).await?; + let sinks = Arc::new(DartUpdateSinkSet::new()); + sinks.register(caller_id.clone(), generation, update_callback_js); + + let veilid_api = match veilid_core::api_startup_json( + update_callback(sinks.clone()), + json_config.clone(), + ) + .await + { + Ok(v) => v, + Err(e) => { + sinks.retire(&caller_id); + return Err(e); + } + }; VEILID_API.replace(Some(veilid_api)); + NODE_STATE.replace(Some(NodeState { + sinks, + config_json: json_config, + })); + VeilidAPIResult::Ok(()) }) } +/// Drop a caller's update sink without closing its stream, for a Dart consumer that cancelled. +#[wasm_bindgen()] +pub fn deregister_update_sink(caller_id: String) { + let sinks = { + let node_state = NODE_STATE.borrow(); + let Some(node_state) = node_state.as_ref() else { + return; + }; + node_state.sinks.clone() + }; + sinks.deregister(&caller_id); +} + #[wasm_bindgen()] pub fn get_veilid_state() -> Promise { wrap_api_future_json(async move { @@ -258,7 +369,13 @@ pub fn is_shutdown() -> VeilidAPIResult { pub fn attach() -> Promise { wrap_api_future_void(async move { let veilid_api = get_veilid_api()?; - veilid_api.attach().await?; + // every caller sharing the node attaches it, so core's "already attached" Generic + // (its only error here other than NotInitialized) is this caller's success + if let Err(e) = veilid_api.attach().await { + if !matches!(e, veilid_core::VeilidAPIError::Generic { .. }) { + return Err(e); + } + } VeilidAPIResult::Ok(()) }) } @@ -267,16 +384,58 @@ pub fn attach() -> Promise { pub fn detach() -> Promise { wrap_api_future_void(async move { let veilid_api = get_veilid_api()?; - veilid_api.detach().await?; + // symmetric with attach: "already detached" is another caller's doing, not an error + if let Err(e) = veilid_api.detach().await { + if !matches!(e, veilid_core::VeilidAPIError::Generic { .. }) { + return Err(e); + } + } VeilidAPIResult::Ok(()) }) } +/// Release `caller_id`'s reference on the running node, tearing it down at zero. #[wasm_bindgen()] -pub fn shutdown_veilid_core() -> Promise { +pub fn shutdown_veilid_core(caller_id: String, generation: String) -> Promise { wrap_api_future_void(async move { + let _lifecycle = LIFECYCLE_LOCK.lock().await; + + let sinks = { + let node_state = NODE_STATE.borrow(); + let Some(node_state) = node_state.as_ref() else { + return Err(veilid_core::VeilidAPIError::NotInitialized); + }; + node_state.sinks.clone() + }; + + // the web stand-in for the FFI dead-port reap: a sink from an earlier Dart program cannot + // be probed, and a detached node can go arbitrarily long without an update to reap it on + sinks.retire_other_generations(&generation); + + // a caller with no registered sink holds no reference + if !sinks.contains(&caller_id) { + return VeilidAPIResult::Ok(()); + } + + if sinks.len() > 1 { + sinks.retire(&caller_id); + return VeilidAPIResult::Ok(()); + } + + // both globals are cleared before the await: the lifecycle lock keeps startup out, but + // every other export runs unlocked and must see a node on its way down as NotInitialized let veilid_api = take_veilid_api()?; + let retired_state = (**NODE_STATE).take(); + + // the last reference stays registered through termination so its consumer sees the + // updates core emits on the way down veilid_api.shutdown().await; + + // unconditional: a teardown that emits no Shutdown update would park every consumer + // in `await for` forever + sinks.close_all(); + drop(retired_state); + VeilidAPIResult::Ok(()) }) } @@ -2142,6 +2301,12 @@ pub fn veilid_version() -> JsValue { ::from_serde(&vv).unwrap_or_log() } +#[wasm_bindgen()] +#[must_use] +pub fn veilid_features() -> JsValue { + to_json(veilid_core::veilid_features()) +} + #[wasm_bindgen()] #[must_use] pub fn default_veilid_config() -> String { diff --git a/veilid-flutter/rust/src/lib.rs b/veilid-flutter/rust/src/lib.rs index 74f6c429..49a0f89f 100644 --- a/veilid-flutter/rust/src/lib.rs +++ b/veilid-flutter/rust/src/lib.rs @@ -17,10 +17,12 @@ cfg_if::cfg_if! { use veilid_core::tools::wasm_tracing::*; use wasm_bindgen_futures::*; + mod dart_js_wrapper; pub mod dart_wasm; pub mod veilid_version; pub mod veilid_wasm_config; + use dart_js_wrapper::*; pub use veilid_version::*; pub use veilid_wasm_config::*; } else { diff --git a/veilid-remote-api/smoketest/Cargo.lock b/veilid-remote-api/smoketest/Cargo.lock index 1962f9f2..a3d858a0 100644 --- a/veilid-remote-api/smoketest/Cargo.lock +++ b/veilid-remote-api/smoketest/Cargo.lock @@ -1491,9 +1491,9 @@ checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" [[package]] name = "hermit-abi" -version = "0.5.2" +version = "0.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" +checksum = "e17592d60ebacc7d5e169f4663c5f84f9161cc90328abcfe8456f41e4dfcb284" [[package]] name = "hex" @@ -3550,9 +3550,9 @@ checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" [[package]] name = "value-bag" -version = "1.13.2" +version = "1.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "068e763e8279de7ab94b6afebded2cb701678af094feb1c12ccb061b4783c1be" +checksum = "11d4788a546b899f8656382fe210e0c478fe3d2dcd6263d50cb6cc268240de6c" [[package]] name = "vcpkg" diff --git a/veilid-tools/smoketest/Cargo.lock b/veilid-tools/smoketest/Cargo.lock index cba16eee..66331df7 100644 --- a/veilid-tools/smoketest/Cargo.lock +++ b/veilid-tools/smoketest/Cargo.lock @@ -715,9 +715,9 @@ checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" [[package]] name = "hermit-abi" -version = "0.5.2" +version = "0.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" +checksum = "e17592d60ebacc7d5e169f4663c5f84f9161cc90328abcfe8456f41e4dfcb284" [[package]] name = "hex" @@ -1922,9 +1922,9 @@ dependencies = [ [[package]] name = "value-bag" -version = "1.13.2" +version = "1.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "068e763e8279de7ab94b6afebded2cb701678af094feb1c12ccb061b4783c1be" +checksum = "11d4788a546b899f8656382fe210e0c478fe3d2dcd6263d50cb6cc268240de6c" [[package]] name = "veilid-hashlink"