restack: remediation architecture + test stack (c0-c3) (#83)

* chore: add tracing migration anchor commit

Apply a deliberately small, mechanical first step by migrating only the fully-finalizable logging call sites and seeding workspace tracing dependencies needed by follow-on commits. This boundary keeps entangled architectural and smoke-oriented rewrites out of the migration anchor so later commits remain reviewable by concern.

Key review files: Cargo.toml, example-streamer-uses/Cargo.toml, utils/integration-test-utils/Cargo.toml, example-streamer-uses/src/bin/common/mod.rs, utils/integration-test-utils/src/integration_test_listeners.rs, utils/integration-test-utils/src/up_client_failing_register.rs, up-streamer/src/endpoint.rs.

Mechanical vs behavioral: mechanical only; logging imports move to tracing where retained, and endpoint startup logging is removed because the final behavior no longer emits those startup logs.

* feat: add structured observability vocabulary module

Introduce a dedicated observability vocabulary leaf module so event names and field keys are centralized before broader domain rewiring. This boundary stays intentionally narrow to give reviewers a fast pass over schema definitions independent of behavior changes.

Key review files: up-streamer/src/observability/events.rs, up-streamer/src/observability/fields.rs, up-streamer/src/observability/mod.rs.

Mechanical vs behavioral: mostly mechanical schema introduction; behavior wiring is intentionally deferred to the architectural commit.

* refactor: extract domain modules and fold subscription-cache into streamer

Collapse the architecture rewrite into one compile-safe boundary that moves route lifecycle, routing, data plane, and runtime ownership behind explicit modules while updating streamer entrypoints/plugins to the new API and tracing setup. This keeps the high-coupling rewire in a single review unit and removes the standalone subscription-cache crate and path dependency in the same step to avoid half-migrated ownership.

Key review files: up-streamer/src/ustreamer.rs, up-streamer/src/control_plane/*, up-streamer/src/data_plane/*, up-streamer/src/routing/*, up-streamer/src/runtime/*, up-streamer/src/lib.rs, up-streamer/Cargo.toml, Cargo.toml, configurable-streamer/src/main.rs, example-streamer-implementations/src/bin/zenoh_someip.rs, up-linux-streamer-plugin/src/lib.rs, utils/usubscription-static-file/Cargo.toml, and minimal compile-safe test API churn in up-streamer/tests/*.

Mechanical vs behavioral: primarily architectural and wiring refactor; behavior-affecting changes are limited to ownership boundaries, add_route API rename alignment, reserved live_usubscription mode fail-fast handling, and mechanical test compile updates required to keep all-target checks green at this commit boundary.

* test: finalize deterministic suite and static subscription behavior

Advance the test-focused phase by finalizing deterministic client-run assertions in up-streamer tests, introducing explicit API contract coverage, and completing integration helper updates while migrating usubscription-static-file to tracing with cache-aware static snapshot behavior. This boundary keeps remaining churn concentrated in tests/helpers and aligns manifests with the planned async-std retirement timing.

Key review files: up-streamer/tests/api_contract_forwarding_rules.rs, up-streamer/tests/support/mod.rs, up-streamer/tests/single_local_*.rs, utils/integration-test-utils/src/integration_test_utils.rs, utils/integration-test-utils/src/up_client_foo.rs, utils/usubscription-static-file/src/lib.rs, utils/integration-test-utils/Cargo.toml, utils/usubscription-static-file/Cargo.toml, Cargo.toml, Cargo.lock.

Mechanical vs behavioral: includes both mechanical import/path migrations and behavioral test determinism updates; overlap on five legacy up-streamer test files is intentional and pre-approved in pr-32-approved-source-overlap.txt as commit-2 gate remediation.

* refactor: align streamer code with up-rust idioms

Adopt up-rust accessors and helpers across runtime code, test utilities, examples, and doctest scaffolding to standardize behavior and reduce manual protobuf traversal.

Retained direct-field usage is intentional where helper equivalents are missing or less efficient.

- Owned authority moves in From<UUri> conversions keep uri.authority_name to avoid extra allocation because authority_name() returns a cloned String.

- Integration test ID regeneration keeps msg.attributes.as_mut() and attributes.id.0 mutation because up-rust 0.9.0 has no mutable ID setter path.

- Full ue_id reads remain where up-rust only exposes split component helpers (uentity_type_id/uentity_instance_id).

- Static subscription canonicalization keeps direct resource_id writes because no setter helper exists.

Author: PLeVasseur

Cargo.lock

@@ -19,7 +19,6 @@ members = [
     "configurable-streamer",
     "up-linux-streamer-plugin",
     "up-streamer",
-    "subscription-cache",
     "utils/usubscription-static-file",
 ]
 
@@ -33,7 +32,6 @@ keywords = ["uProtocol", "SDV", "routing", "streamer"]
 license = "Apache-2.0"
 
 [workspace.dependencies]
-async-std = { version = "1.12.0", features = ["attributes"] }
 async-trait = { version = "0.1" }
 clap = { version = "4.5" }
 env_logger = { version = "0.10.1" }
@@ -43,6 +41,8 @@ log = { version = "0.4.20" }
 json5 = { version = "0.4.1" }
 serde = { version = "1.0.154", features = ["derive"] }
 serde_json = { version = "1.0.94" }
+tracing = { version = "0.1" }
+tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
 uuid = { version = "1.7.0" }
 tokio = { version = "1.44", default-features = false, features = [
     "rt",

Cargo.toml

@@ -16,6 +16,8 @@
       ue_version_major: 1
     },
     usubscription_config: {
+      // Provider mode: static_file (default) or reserved live_usubscription
+      mode: "static_file",
       // Lists the path to the subscription file when using static file
       file_path: "subscription_data.json"
     },

configurable-streamer/CONFIG.json5

@@ -22,12 +22,12 @@ license.workspace = true
 [dependencies]
 async-trait = { workspace = true }
 clap = { workspace = true }
-env_logger = { workspace = true }
-log = { workspace = true }
 json5 = { workspace = true }
 protobuf = { workspace = true }
 serde = { workspace = true }
-tokio = { workspace = true }
+tokio = { workspace = true, features = ["signal"] }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
 up-rust = { workspace = true }
 up-streamer = { path = "../up-streamer" }
 up-transport-zenoh = { workspace = true }

configurable-streamer/Cargo.toml

@@ -39,9 +39,19 @@ pub struct StreamerUuri {
 #[derive(Deserialize, Serialize, Debug, Clone)]
 #[serde(deny_unknown_fields)]
 pub struct USubscriptionConfig {
+    #[serde(default)]
+    pub(crate) mode: SubscriptionProviderMode,
     pub(crate) file_path: String,
 }
 
+#[derive(Deserialize, Serialize, Debug, Clone, Default)]
+#[serde(rename_all = "snake_case")]
+pub enum SubscriptionProviderMode {
+    #[default]
+    StaticFile,
+    LiveUsubscription,
+}
+
 #[derive(Deserialize, Serialize, Debug, Clone)]
 #[serde(deny_unknown_fields)]
 pub struct Transports {

configurable-streamer/src/config.rs

@@ -13,13 +13,13 @@
 
 mod config;
 
-use crate::config::Config;
+use crate::config::{Config, EndpointConfig, SubscriptionProviderMode};
 use clap::Parser;
-use log::info;
 use std::io::Read;
 use std::sync::Arc;
-use std::thread;
 use std::{collections::HashMap, fs::File};
+use tracing::info;
+use up_rust::core::usubscription::USubscription;
 use up_rust::{UCode, UStatus, UTransport};
 use up_streamer::{Endpoint, UStreamer};
 use up_transport_mqtt5::{Mqtt5Transport, Mqtt5TransportOptions, MqttClientOptions};
@@ -33,9 +33,79 @@ struct StreamerArgs {
     config: String,
 }
 
+fn register_transport_endpoints(
+    endpoints: &mut HashMap<String, Endpoint>,
+    endpoint_configs: &[EndpointConfig],
+    transport: Arc<dyn UTransport>,
+) -> Result<(), UStatus> {
+    for endpoint_config in endpoint_configs {
+        let endpoint = Endpoint::new(
+            &endpoint_config.endpoint,
+            &endpoint_config.authority,
+            transport.clone(),
+        );
+
+        if endpoints
+            .insert(endpoint_config.endpoint.clone(), endpoint)
+            .is_some()
+        {
+            return Err(UStatus::fail_with_code(
+                UCode::INVALID_ARGUMENT,
+                format!(
+                    "Duplicate endpoint name found: {}",
+                    endpoint_config.endpoint
+                ),
+            ));
+        }
+    }
+
+    Ok(())
+}
+
+async fn wire_forwarding_rules(
+    streamer: &mut UStreamer,
+    endpoints: &HashMap<String, Endpoint>,
+    endpoint_configs: &[EndpointConfig],
+) -> Result<(), UStatus> {
+    for endpoint_config in endpoint_configs {
+        for forwarding_target in &endpoint_config.forwarding {
+            let left_endpoint = endpoints.get(&endpoint_config.endpoint).ok_or_else(|| {
+                UStatus::fail_with_code(
+                    UCode::INVALID_ARGUMENT,
+                    format!(
+                        "Unknown endpoint in forwarding rules: {}",
+                        endpoint_config.endpoint
+                    ),
+                )
+            })?;
+            let right_endpoint = endpoints.get(forwarding_target).ok_or_else(|| {
+                UStatus::fail_with_code(
+                    UCode::INVALID_ARGUMENT,
+                    format!("Unknown forwarding target endpoint: {forwarding_target}"),
+                )
+            })?;
+
+            streamer
+                .add_route_ref(left_endpoint, right_endpoint)
+                .await?;
+        }
+    }
+
+    Ok(())
+}
+
+async fn wait_for_shutdown_signal() -> Result<(), UStatus> {
+    tokio::signal::ctrl_c().await.map_err(|error| {
+        UStatus::fail_with_code(
+            UCode::INTERNAL,
+            format!("Unable to wait for shutdown signal: {error:?}"),
+        )
+    })
+}
+
 #[tokio::main]
 async fn main() -> Result<(), UStatus> {
-    env_logger::init();
+    let _ = tracing_subscriber::fmt::try_init();
 
     info!("Started up-linux-streamer-configurable");
 
@@ -57,49 +127,71 @@ async fn main() -> Result<(), UStatus> {
             format!("Unable to parse config file: {e:?}"),
         )
     })?;
-    config.transports.mqtt.load_mqtt_details().unwrap();
+    config.transports.mqtt.load_mqtt_details().map_err(|e| {
+        UStatus::fail_with_code(
+            UCode::INVALID_ARGUMENT,
+            format!("Unable to load MQTT transport details: {e:?}"),
+        )
+    })?;
 
-    let subscription_path = config.usubscription_config.file_path;
-    let usubscription = Arc::new(USubscriptionStaticFile::new(subscription_path));
+    let usubscription: Arc<dyn USubscription> = match config.usubscription_config.mode {
+        SubscriptionProviderMode::StaticFile => Arc::new(USubscriptionStaticFile::new(
+            config.usubscription_config.file_path.clone(),
+        )),
+        SubscriptionProviderMode::LiveUsubscription => {
+            return Err(UStatus::fail_with_code(
+                    UCode::UNIMPLEMENTED,
+                    "live_usubscription mode is reserved in this phase; live runtime integration is deferred (see reports/usubscription-decoupled-pubsub-migration/05-live-integration-deferred.md)",
+                ));
+        }
+    };
 
     // Start the streamer instance.
     let mut streamer = UStreamer::new(
         "up-streamer",
         config.up_streamer_config.message_queue_size,
         usubscription,
     )
-    .expect("Failed to create uStreamer");
+    .await?;
 
     let mut endpoints: HashMap<String, Endpoint> = HashMap::new();
 
     // build the zenoh transport
-    let zenoh_config = ZenohConfig::from_file(config.transports.zenoh.config_file).unwrap();
+    let zenoh_config =
+        ZenohConfig::from_file(config.transports.zenoh.config_file).map_err(|e| {
+            UStatus::fail_with_code(
+                UCode::INVALID_ARGUMENT,
+                format!("Unable to load Zenoh config file: {e:?}"),
+            )
+        })?;
     let zenoh_transport: Arc<dyn UTransport> = Arc::new(
         UPTransportZenoh::builder(config.streamer_uuri.authority.clone())
-            .expect("Unable to create Zenoh transport builder")
+            .map_err(|e| {
+                UStatus::fail_with_code(
+                    UCode::INTERNAL,
+                    format!("Unable to create Zenoh transport builder: {e:?}"),
+                )
+            })?
             .with_config(zenoh_config)
             .build()
             .await
-            .expect("Unable to initialize Zenoh UTransport"),
+            .map_err(|e| {
+                UStatus::fail_with_code(
+                    UCode::INTERNAL,
+                    format!("Unable to initialize Zenoh UTransport: {e:?}"),
+                )
+            })?,
     );
 
     // build the mqtt5 transport
+    let mqtt_details = config.transports.mqtt.mqtt_details.clone().ok_or_else(|| {
+        UStatus::fail_with_code(
+            UCode::INVALID_ARGUMENT,
+            "MQTT transport details are missing after load_mqtt_details",
+        )
+    })?;
     let mqtt_client_options = MqttClientOptions {
-        broker_uri: config
-            .transports
-            .mqtt
-            .mqtt_details
-            .clone()
-            .unwrap()
-            .hostname
-            + ":"
-            + &config
-                .transports
-                .mqtt
-                .mqtt_details
-                .unwrap()
-                .port
-                .to_string(),
+        broker_uri: format!("{}:{}", mqtt_details.hostname, mqtt_details.port),
         ..Default::default()
     };
     let mqtt_transport_options = Mqtt5TransportOptions {
@@ -114,73 +206,29 @@ async fn main() -> Result<(), UStatus> {
     mqtt5_transport.connect().await?;
     let mqtt5_transport: Arc<dyn UTransport> = Arc::new(mqtt5_transport);
 
-    // build all zenoh endpoints
-    for zenoh_endpoint_config in config.transports.zenoh.endpoints.clone() {
-        let endpoint = Endpoint::new(
-            &zenoh_endpoint_config.endpoint,
-            &zenoh_endpoint_config.authority,
-            zenoh_transport.clone(),
-        );
-        if endpoints
-            .insert(zenoh_endpoint_config.endpoint.clone(), endpoint)
-            .is_some()
-        {
-            return Err(UStatus::fail_with_code(
-                UCode::INVALID_ARGUMENT,
-                format!(
-                    "Duplicate endpoint name found: {}",
-                    zenoh_endpoint_config.endpoint
-                ),
-            ));
-        }
-    }
-
-    // build all mqtt endpoints
-    for mqtt_endpoint_config in config.transports.mqtt.endpoints.clone() {
-        let endpoint = Endpoint::new(
-            &mqtt_endpoint_config.endpoint,
-            &mqtt_endpoint_config.authority,
-            mqtt5_transport.clone(),
-        );
-        if endpoints
-            .insert(mqtt_endpoint_config.endpoint.clone(), endpoint)
-            .is_some()
-        {
-            return Err(UStatus::fail_with_code(
-                UCode::INVALID_ARGUMENT,
-                format!(
-                    "Duplicate endpoint name found: {}",
-                    mqtt_endpoint_config.endpoint
-                ),
-            ));
-        }
-    }
-
-    // set up the endpoint forwarding for zenoh
-    for zenoh_endpoint in config.transports.zenoh.endpoints {
-        for forwarding in zenoh_endpoint.forwarding {
-            let left_endpoint = endpoints.get(&zenoh_endpoint.endpoint).unwrap();
-            let right_endpoint = endpoints.get(&forwarding).unwrap();
-            streamer
-                .add_forwarding_rule(left_endpoint.to_owned(), right_endpoint.to_owned())
-                .await
-                .expect("Could not add forwarding rule from {zenoh.endpoint} to {forwarding}");
-        }
-    }
-
-    // set up the endpoint forwarding for mqtt
-    for mqtt5_endpoint in config.transports.mqtt.endpoints {
-        for forwarding in mqtt5_endpoint.forwarding {
-            let left_endpoint = endpoints.get(&mqtt5_endpoint.endpoint).unwrap();
-            let right_endpoint = endpoints.get(&forwarding).unwrap();
-            streamer
-                .add_forwarding_rule(left_endpoint.to_owned(), right_endpoint.to_owned())
-                .await
-                .expect("Could not add forwarding rule from {mqtt.endpoint} to {forwarding}");
-        }
-    }
+    register_transport_endpoints(
+        &mut endpoints,
+        &config.transports.zenoh.endpoints,
+        zenoh_transport,
+    )?;
+    register_transport_endpoints(
+        &mut endpoints,
+        &config.transports.mqtt.endpoints,
+        mqtt5_transport,
+    )?;
+
+    wire_forwarding_rules(
+        &mut streamer,
+        &endpoints,
+        &config.transports.zenoh.endpoints,
+    )
+    .await?;
+    wire_forwarding_rules(&mut streamer, &endpoints, &config.transports.mqtt.endpoints).await?;
 
-    thread::park();
+    println!("READY streamer_initialized");
+    info!("Streamer initialized; waiting for shutdown signal");
+    wait_for_shutdown_signal().await?;
+    info!("Shutdown signal received; exiting");
 
     Ok(())
 }

configurable-streamer/src/main.rs

@@ -33,12 +33,12 @@ bundled-vsomeip = ["up-transport-vsomeip/bundled"]
 async-trait = { workspace = true }
 chrono = { version = "0.4" }
 clap = { version = "4.5.9", features = ["derive"] }
-env_logger = { version = "0.10.2" }
-log = { workspace = true }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
 json5 = { workspace = true }
 protobuf = { workspace = true }
 serde = { workspace = true }
-tokio = { workspace = true }
+tokio = { workspace = true, features = ["signal"] }
 up-rust = { workspace = true }
 up-streamer = { path = "../up-streamer" }
 up-transport-zenoh = { workspace = true, optional = true }

example-streamer-implementations/Cargo.toml

@@ -20,6 +20,8 @@
       ue_version_major: 1
     },
     usubscription_config: {
+      // Provider mode: static_file (default) or reserved live_usubscription
+      mode: "static_file",
       // Lists the path to the subscription file when using static file
       file_path: "../example-streamer-implementations/subscription_data.json"
     },