veilid-flutter multi-attach fixes

This commit is contained in:
Christien Rioux committed 2026-08-31 12:49:22 -04:00
1 parent 8f39c997ab
commit 9d065d70a5
21 files changed
+1195 -187

No files matched your search

+1
View File
@@ -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
+4 -4
View File
@@ -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"
+5
View File
@@ -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<K: AsRef<str>>(
&self,
context: ProtectedStoreContext,
+36 -15
View File
@@ -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<ProtectedStoreBackend> {
let config = ps.config();
@@ -139,15 +151,24 @@ pub(super) async fn build_backend(ps: &ProtectedStore) -> VeilidAPIResult<Protec
let mut keyring_manager: Option<KeyringManager> = 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
+20 -2
View File
@@ -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::<jni::errors::ThrowRuntimeExAndDefault>();
}
/// 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<JObject<'static>>) {
let globals_locked = ANDROID_GLOBALS.lock();
let globals = globals_locked.as_ref().unwrap_or_log();
+1
View File
@@ -4,6 +4,7 @@ concurrency: 1
tags:
config: {}
startup: {}
crypto: {}
table_db: {}
routing_context: {}
@@ -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']);
});
});
@@ -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<VeilidConfig> _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<VeilidUpdate> 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<void>();
final _attachment = Completer<VeilidUpdateAttachment>();
late final StreamSubscription<VeilidUpdate> _subscription;
Completer<void>? _pending;
int _pendingCount = 0;
Future<void> get done => _done.future;
Future<VeilidUpdateAttachment> get firstAttachment => _attachment.future;
bool get isDone => _done.isCompleted;
/// Completes once [count] more updates have arrived.
Future<void> nextUpdates(int count) {
final pending = _pending = Completer<void>();
_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<void> cancel() => _subscription.cancel();
}
class TestStartupShutdown {
late final VeilidConfig config;
late final VeilidConfig otherConfig;
Future<void> 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<void> 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<void> 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<void> 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<VeilidUpdateAttachment>());
expect(await second.firstAttachment.timeout(_updateTimeout),
isA<VeilidUpdateAttachment>());
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<void> testFailedStartupTakesNoReference() async {
final invalidConfig = config.copyWith(programName: '$_programName/invalid');
await expectLater(() => Veilid.instance.startupVeilidCore(invalidConfig),
throwsA(isA<VeilidAPIException>()));
// 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<void> testDifferentInitKeyIsRejected() async {
final stream = await Veilid.instance.startupVeilidCore(config);
final collector = _UpdateCollector(stream);
await expectLater(() => Veilid.instance.startupVeilidCore(otherConfig),
throwsA(isA<VeilidAPIExceptionAlreadyInitialized>()));
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<void> testExtraShutdownsDoNotUnderflow() async {
await expectLater(Veilid.instance.shutdownVeilidCore,
throwsA(isA<VeilidAPIExceptionNotInitialized>()));
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<VeilidAPIExceptionNotInitialized>()));
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<void> 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<VeilidAPIExceptionNotInitialized>()));
}
}
+6 -2
View File
@@ -57,7 +57,8 @@ class _MyAppState extends State<MyApp> 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<MyApp> 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) {
+20 -11
View File
@@ -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<void> 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<void> 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.
+34 -10
View File
@@ -61,7 +61,7 @@ void _eagerResolveNativeSymbols() {
Native.addressOf<NativeFunction<Void Function(Int64)>>(VeilidFFI._getVeilidState),
Native.addressOf<NativeFunction<Void Function(Int64)>>(VeilidFFI._attach),
Native.addressOf<NativeFunction<Void Function(Int64)>>(VeilidFFI._detach),
Native.addressOf<NativeFunction<Void Function(Int64)>>(VeilidFFI._shutdownVeilidCore),
Native.addressOf<NativeFunction<Void Function(Int64, Int64)>>(VeilidFFI._shutdownVeilidCore),
Native.addressOf<NativeFunction<Void Function(Int64)>>(VeilidFFI._routingContext),
Native.addressOf<NativeFunction<Int32 Function(Uint32)>>(VeilidFFI._releaseRoutingContext),
Native.addressOf<NativeFunction<Uint32 Function(Uint32)>>(VeilidFFI._routingContextWithDefaultSafety),
@@ -409,7 +409,12 @@ Stream<T> processStreamJson<T>(
) async* {
try {
await for (final value in port) {
final list = value as List<dynamic>;
// 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<dynamic>) {
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 = <int>[];
// Shared library functions
@Native<Void Function(Pointer<Utf8>)>(symbol: 'free_string')
external static void _freeString(Pointer<Utf8> s);
@@ -1750,8 +1760,8 @@ class VeilidFFI extends Veilid {
@Native<Void Function(Int64)>(symbol: 'detach')
external static void _detach(int port);
@Native<Void Function(Int64)>(symbol: 'shutdown_veilid_core')
external static void _shutdownVeilidCore(int port);
@Native<Void Function(Int64, Int64)>(symbol: 'shutdown_veilid_core')
external static void _shutdownVeilidCore(int port, int streamPort);
@Native<Void Function(Int64)>(symbol: 'routing_context')
external static void _routingContext(int port);
@@ -2544,7 +2554,7 @@ class VeilidFFI extends Veilid {
}
@override
Future<Stream<VeilidUpdate>> startupVeilidCore(VeilidConfig config) {
Future<Stream<VeilidUpdate>> 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<void> 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);
}
+85 -20
View File
@@ -40,6 +40,9 @@ String _wasmCallString(String method, [List<JSAny?>? arguments]) =>
int _wasmCallInt(String method, [List<JSAny?>? arguments]) =>
wasm.callMethodVarArgs<JSNumber>(method.toJS, arguments).toDartInt;
bool _wasmCallBool(String method, [List<JSAny?>? arguments]) =>
wasm.callMethodVarArgs<JSBoolean>(method.toJS, arguments).toDart;
void _wasmCallVoid(String method, [List<JSAny?>? 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 = <String>[];
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<VeilidUpdate> 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<String, dynamic>,
),
);
} on Object catch (e) {
web.console.error('veilid: dropped an undecodable update: $e'.toJS);
}
}).toJS;
@override
void initializeVeilidCore(Map<String, dynamic> platformConfigJson) {
final platformConfigJsonString = jsonEncode(platformConfigJson);
@@ -1018,31 +1061,42 @@ class VeilidJS extends Veilid {
@override
Future<Stream<VeilidUpdate>> startupVeilidCore(VeilidConfig config) async {
final streamController = StreamController<VeilidUpdate>();
void updateCallback(String update) {
final updateJson = jsonDecode(update) as Map<String, dynamic>;
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<VeilidUpdate> sink;
sink = StreamController<VeilidUpdate>(
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<VeilidState> 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<bool> isShutdown() => _wasmPromiseBool('is_shutdown', []);
Future<bool> isShutdown() async => _wasmCallBool('is_shutdown', []);
@override
Future<void> attach() => _wasmPromiseVoid('attach', []);
@@ -1051,8 +1105,16 @@ class VeilidJS extends Veilid {
Future<void> detach() => _wasmPromiseVoid('detach', []);
@override
Future<void> shutdownVeilidCore() =>
_wasmPromiseVoid('shutdown_veilid_core', []);
Future<void> 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<CryptoKind> validCryptoKinds() => _wasmCallJson(
@@ -1249,8 +1311,11 @@ class VeilidJS extends Veilid {
}
@override
List<String> veilidFeatures() =>
_wasmCallJson('veilid_features', (x) => x as List<String>, []);
List<String> veilidFeatures() => _wasmCallJson(
'veilid_features',
(x) => (x as List<dynamic>).map((e) => e as String).toList(),
[],
);
@override
String defaultVeilidConfig() => _wasmCallString('default_veilid_config', []);
@@ -293,13 +293,16 @@ class DefaultVeilidFixture implements VeilidFixture {
Future<void> 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();
}
}
}
+2 -1
View File
@@ -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"
+138 -23
View File
@@ -121,6 +121,7 @@ lazy_static! {
static ref DHT_TRANSACTIONS: Mutex<BTreeMap<u32, DHTTransaction>> = Mutex::new(BTreeMap::new());
static ref FLAME_GUARD: Mutex<Option<tracing_flame::FlushGuard<std::io::BufWriter<std::fs::File>>>> =
Mutex::new(None);
static ref NODE_STATE: Mutex<Option<NodeState>> = Mutex::new(None);
}
async fn get_veilid_api() -> VeilidAPIResult<VeilidAPI> {
@@ -131,6 +132,54 @@ async fn get_veilid_api() -> VeilidAPIResult<VeilidAPI> {
.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<DartUpdateSinkSet>,
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<InitKey> {
let key = veilid_core::deserialize_json::<VeilidFFIConfigInitKey>(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<DartUpdateSinkSet>) -> 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(),
+194 -66
View File
@@ -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<Isolate>,
/// 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<Mutex<DartIsolateStreamInner>>,
}
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<T: IntoDart>(&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<T: Serialize + Debug>(&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<E: IntoDart + Debug>(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<E: Serialize + Debug>(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<BTreeMap<i64, DartUpdateSink>>,
}
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<T: Serialize + Debug>(&self, value: T) {
let sinks: Vec<DartUpdateSink> = self.sinks.lock().values().copied().collect();
if sinks.is_empty() {
return;
}
let json = veilid_core::serialize_json(value);
let mut dead = Vec::<i64>::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<i64> = self
.sinks
.lock()
.keys()
.copied()
.filter(|port| *port != keep)
.collect();
let dead: Vec<i64> = 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));
}
}
+173
View File
@@ -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<RefCell<BTreeMap<String, DartUpdateSink>>>,
}
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<DartUpdateSink> = {
let mut sinks = self.sinks.borrow_mut();
let stale_ids: Vec<String> = 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<T: Serialize + fmt::Debug>(&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::<String>::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<DartUpdateSink> = {
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();
}
}
}
+183 -18
View File
@@ -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<RefCell<BTreeMap<u32, veilid_core::TableDBTransaction>>> =
SendWrapper::new(RefCell::new(BTreeMap::new()));
static ref NODE_STATE: SendWrapper<RefCell<Option<NodeState>>> =
SendWrapper::new(RefCell::new(None));
static ref LIFECYCLE_LOCK: AsyncMutex<()> = AsyncMutex::new(());
}
fn get_veilid_api() -> Result<veilid_core::VeilidAPI, veilid_core::VeilidAPIError> {
@@ -42,6 +46,56 @@ fn take_veilid_api() -> Result<veilid_core::VeilidAPI, veilid_core::VeilidAPIErr
.ok_or(veilid_core::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 sink whose Dart context died is not a shutdown.
///
/// The init key is deliberately not cached here. A Dart hot restart discards the Dart side while
/// this module and the node it holds survive, so the join decision reads the running node's own
/// config instead of anything either side remembered.
struct NodeState {
sinks: Arc<DartUpdateSinkSet>,
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<InitKey> {
let key = veilid_core::deserialize_json::<VeilidWASMConfigInitKey>(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<InitKey> {
let config = veilid_api.config()?;
Ok((config.program_name.clone(), config.namespace.clone()))
}
fn update_callback(sinks: Arc<DartUpdateSinkSet>) -> 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<T: Serialize + fmt::Debug>(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<bool> {
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 {
<JsValue as JsValueSerdeExt>::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 {
+2
View File
@@ -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 {
+4 -4
View File
@@ -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"
+4 -4
View File
@@ -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"