feat(fleet): honor local transport order for replication and peer relay

crates/fractal-peer/src/peer.rs

@@ -322,6 +322,8 @@ pub struct ClientRelay {
     /// Reuse sealed membership across per-item reachability checks for the same short TTL.
     pub membership: Arc<std::sync::Mutex<Option<ObservedMembership>>>,
     pub fabric: Option<Fabric>,
+    /// Ordered local transport policy; absent for a legacy config-only fleet.
+    transports: Option<Vec<String>>,
     legacy: bool,
     /// When each owner last refused this node's reads as stale, for provenance.
     fence_conflicts:
@@ -433,6 +435,25 @@ impl ClientRelay {
     /// peer, then the peers with the shortest observed path to it. With no observation of the
     /// target at all, every peer is worth a try. Nodes the read already passed are never chosen.
     pub fn next_hops(&self, target: &str, visited: &[String]) -> Vec<PeerConfig> {
+        self.next_hops_with_transports(
+            target,
+            visited,
+            LocalTransports {
+                fabric: self.fabric.is_some(),
+                tailscale: local_addresses()
+                    .iter()
+                    .any(crate::fleet::transport::is_tailnet_address),
+                ..LocalTransports::default()
+            },
+        )
+    }
+
+    fn next_hops_with_transports(
+        &self,
+        target: &str,
+        visited: &[String],
+        available: LocalTransports,
+    ) -> Vec<PeerConfig> {
         let view = self.fleet_view();
         // Sync from a dial-out member supplies no reverse owner-RPC or PTY transport.
         // In particular, dropping its old links must not enable the unseen-owner fallback.
@@ -441,12 +462,11 @@ impl ClientRelay {
         }) {
             return Vec::new();
         }
-        let local = LocalTransports {
-            fabric: self.fabric.is_some(),
-            tailscale: local_addresses()
-                .iter()
-                .any(crate::fleet::transport::is_tailnet_address),
-        };
+        let local = LocalTransports::configured(
+            self.transports.as_deref(),
+            available.tailscale,
+            available.fabric,
+        );
         let dialable = dial_targets(&view, &self.node, &self.peers, local);
         let names = dialable
             .keys()
@@ -512,14 +532,21 @@ impl ClientRelay {
             observed: Arc::default(),
             membership: Arc::default(),
             fence_conflicts: Arc::default(),
-            fabric: resolve_tool(
-                config
-                    .fleet
-                    .as_ref()
-                    .and_then(|file| file.fabric.as_deref()),
-                "fabric",
-            )
-            .map(Fabric::new),
+            transports: config.fleet.as_ref().map(|file| file.transports.clone()),
+            fabric: if config
+                .fleet
+                .as_ref()
+                .is_none_or(|file| file.transports.iter().any(|name| name == "fabric"))
+                || config.peers.iter().any(|peer| peer.url.starts_with("fabric://"))
+            {
+                resolve_tool(
+                    config.fleet.as_ref().and_then(|file| file.fabric.as_deref()),
+                    "fabric",
+                )
+                .map(Fabric::new)
+            } else {
+                None
+            },
         }))
     }
 
@@ -2034,6 +2061,114 @@ mod tests {
 
 
 
+    #[test]
+    fn relay_next_hops_follow_local_transport_order_and_removal_with_legacy_peers() {
+        use serde_json::json;
+
+        let view: FleetView = serde_json::from_value(json!({
+            "members": [{
+                "name": "owner",
+                "member_key": "owner-key",
+                "state": "current",
+                "mode": "listening",
+                "start": 1,
+                // Old peers advertise endpoints without an ordering or preference field.
+                "endpoints": [
+                    {"transport": "loopback", "address": "127.0.0.1:4000"},
+                    {"transport": "tailscale", "address": "100.101.102.103:4000"},
+                    {"transport": "fabric", "node": "owner-node", "protocol": "st3-peer-v1"}
+                ]
+            }]
+        })).unwrap();
+        let mut relay = ClientRelay {
+            node: "gateway".into(),
+            peers: Vec::new(),
+            auth: FleetAuth::test("fleet-test", &[7; 32]),
+            http: reqwest::Client::new(),
+            links: Some(Arc::new(Store::open_memory("gateway").unwrap())),
+            observed: Arc::default(),
+            membership: Arc::new(std::sync::Mutex::new(Some((
+                std::time::Instant::now(),
+                Arc::new(view),
+            )))),
+            fabric: None,
+            transports: None,
+            legacy: false,
+            fence_conflicts: Arc::default(),
+        };
+        let available = LocalTransports {
+            tailscale: true,
+            fabric: true,
+            ..LocalTransports::default()
+        };
+        let tailscale = "http://100.101.102.103:4000";
+        let fabric = "fabric://owner-node/st3-peer-v1";
+        let loopback = "http://127.0.0.1:4000";
+        for (policy, expected) in [
+            (None, vec![tailscale, fabric, loopback]),
+            (Some(vec!["tailscale", "fabric"]), vec![tailscale, fabric, loopback]),
+            (Some(vec!["fabric", "tailscale"]), vec![fabric, tailscale, loopback]),
+            (Some(vec!["fabric"]), vec![fabric, loopback]),
+            (Some(vec!["tailscale"]), vec![tailscale, loopback]),
+            (Some(Vec::new()), vec![loopback]),
+        ] {
+            relay.transports = policy.map(|names| names.into_iter().map(str::to_owned).collect());
+            assert_eq!(
+                relay.next_hops_with_transports("owner", &[], available)
+                    .into_iter()
+                    .map(|peer| peer.url)
+                    .collect::<Vec<_>>(),
+                expected,
+                "{:?}",
+                relay.transports,
+            );
+        }
+        // A daemon restart reconstructs the relay from the saved fleet file. Preference
+        // and removal must affect real next hops, not just the serialized setting.
+        let root = tempfile::tempdir().unwrap();
+        let secret = root.path().join("fleet-secret");
+        fs::write(&secret, [7_u8; 32]).unwrap();
+        fs::set_permissions(&secret, fs::Permissions::from_mode(0o600)).unwrap();
+        let key = root.path().join("node.key");
+        MemberKey::load_or_create(&key).unwrap();
+        let mut file = smallclaims::fleet::FleetFile {
+            fleet_id: "1f91ca65-7793-48cc-866e-ac15690130e1".into(),
+            node_key_file: key,
+            ..Default::default()
+        };
+        for (policy, expected) in [
+            (vec!["tailscale", "fabric"], vec![tailscale, fabric, loopback]),
+            (vec!["fabric", "tailscale"], vec![fabric, tailscale, loopback]),
+            (vec!["tailscale"], vec![tailscale, loopback]),
+        ] {
+            file.transports = policy.into_iter().map(str::to_owned).collect();
+            file.save(root.path()).unwrap();
+            let restarted = ClientRelay::from_config(&Config {
+                node: "gateway".into(),
+                fleet_id: Some(file.fleet_id.clone()),
+                shared_secret_file: Some(secret.clone()),
+                fleet: smallclaims::fleet::FleetFile::load(root.path()).unwrap(),
+                state_dir: root.path().into(),
+                ..Default::default()
+            }).unwrap().unwrap().with_links(relay.links.as_ref().unwrap().clone());
+            *restarted.membership.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some((
+                std::time::Instant::now(),
+                relay.fleet_view(),
+            ));
+            assert_eq!(
+                restarted.next_hops_with_transports("owner", &[], available)
+                    .into_iter().map(|peer| peer.url).collect::<Vec<_>>(),
+                expected,
+            );
+        }
+        // Explicit config routes retain their precedence for mixed config/member fleets.
+        relay.peers.push(PeerConfig {
+            name: "owner".into(),
+            url: fabric.into(),
+        });
+        assert_eq!(relay.next_hops_with_transports("owner", &[], available)[0].url, fabric);
+    }
+
     #[test]
     fn next_hops_prefer_the_owner_then_the_nearest_peer_that_reaches_it() {
         let links = |pairs: &[(&str, &str)]| {

crates/fractal/src/fleet_cli.rs

@@ -459,7 +459,7 @@ pub(super) async fn run_fleet(endpoint: &Endpoint, command: FleetCommand, json_o
                 "This member is now {}; it announces the change when its replication worker starts.",
                 file.mode.as_str()
             );
-            refresh_fleet_worker(args.no_service)?;
+            refresh_fleet_services(args.no_service)?;
             Ok(())
         }
         FleetCommand::Transports { command } => {
@@ -648,15 +648,15 @@ async fn save_and_publish_fleet_settings(
             },
         )
         .await
-        .context("fleet.toml saved; endpoint publication did not complete; restart the replication worker to announce the saved settings")?;
+        .context("fleet.toml saved; endpoint publication did not complete; restart st3 daemon and st3 replication-worker to apply saved relay settings and announce the saved endpoints")?;
     Ok(())
 }
 
-fn refresh_fleet_worker(no_service: bool) -> Result<()> {
+fn refresh_fleet_services(no_service: bool) -> Result<()> {
     if !no_service && services_installed()? {
         fractal_daemon::service::install(Config::load_with_fleet(None)?)?;
     } else {
-        println!("Restart st3 replication-worker for the change to take effect.");
+        println!("Restart st3 daemon and st3 replication-worker for the change to take effect.");
         println!(
             "The old worker no longer publishes endpoints after its settings change; restart it to resume publication."
         );
@@ -685,8 +685,20 @@ async fn run_fleet_transports(
         .transports
         .iter()
         .any(|configured| configured == transport);
-    // A repeated add never reconfigures an existing exposure, even with a different flag.
-    if matches!(&command, FleetTransportsCommand::Add { .. }) && enabled {
+    // Re-adding can change local preference, but never reconfigures the exposure.
+    if let FleetTransportsCommand::Add {
+        prefer, no_service, ..
+    } = &command
+        && enabled
+    {
+        if *prefer && file.transports.first() != Some(transport) {
+            file.transports.retain(|configured| configured != transport);
+            file.transports.insert(0, transport.clone());
+            file.save(&config.state_dir)?;
+            drop(lock);
+            println!("Fabric is now preferred; its existing protocol is unchanged.");
+            return refresh_fleet_services(*no_service);
+        }
         println!("Fabric is already enabled; remove it before changing its protocol.");
         return Ok(());
     }
@@ -707,6 +719,7 @@ async fn run_fleet_transports(
     let no_service = match command {
         FleetTransportsCommand::Add {
             fabric_protocol,
+            prefer,
             no_service,
             ..
         } => {
@@ -738,7 +751,11 @@ async fn run_fleet_transports(
                 .expose(&protocol, &format!("127.0.0.1:{port}"))
                 .await
                 .context("Fabric exposure failed; fleet settings and endpoints are unchanged")?;
-            file.transports.push("fabric".into());
+            if prefer {
+                file.transports.insert(0, "fabric".into());
+            } else {
+                file.transports.push("fabric".into());
+            }
             file.fabric_protocol = Some(protocol.clone());
             endpoints.retain(|endpoint| endpoint["transport"] != "fabric");
             endpoints.push(json!({"transport":"fabric", "node":node, "protocol":protocol}));
@@ -790,8 +807,8 @@ async fn run_fleet_transports(
                 None => Err(anyhow::anyhow!("fabric is not installed")),
             };
             drop(lock);
-            // Refresh even when cleanup fails: no worker may keep using the removed carrier.
-            refresh_fleet_worker(no_service)?;
+            // Refresh even when cleanup fails: relay and worker must stop using the carrier.
+            refresh_fleet_services(no_service)?;
             if let Err(error) = result {
                 println!(
                     "Fabric endpoint removed; unexposure still needs cleanup: {error:#}. When Fabric is available, run fabric unexpose {protocol}"
@@ -805,7 +822,7 @@ async fn run_fleet_transports(
         }
     };
     drop(lock);
-    refresh_fleet_worker(no_service)
+    refresh_fleet_services(no_service)
 }
 
 /// Settings for a migrating node: its existing replication port unless one is given.

crates/fractal/src/main.rs

@@ -930,15 +930,18 @@ enum FleetTransportsCommand {
         /// Use this protocol instead of the saved protocol or st3/fleet/FLEET_ID.
         #[arg(long)]
         fabric_protocol: Option<String>,
-        /// Leave the replication worker restart to the operator.
+        /// Prefer Fabric over other automatic routes, including when already enabled.
+        #[arg(long)]
+        prefer: bool,
+        /// Leave the daemon and replication worker restarts to the operator.
         #[arg(long)]
         no_service: bool,
     },
     /// Withdraw Fabric's endpoint, then unexpose it. Keep at least one usable transport.
     Remove {
         #[arg(value_parser = ["fabric"])]
         transport: String,
-        /// Leave the replication worker restart to the operator.
+        /// Leave the daemon and replication worker restarts to the operator.
         #[arg(long)]
         no_service: bool,
     },

crates/fractal/tests/fleet.rs

@@ -3640,6 +3640,70 @@ async fn an_existing_member_adds_and_removes_fabric_without_rejoining() {
     }
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn fabric_transport_preference_preserves_other_transports_and_existing_exposure() {
+    if fractal_daemon::test_support::supervise_test() {
+        return;
+    }
+    let root = tempfile::tempdir().unwrap();
+    for prefer_on_add in [false, true] {
+        let name = if prefer_on_add { "preferred" } else { "appended" };
+        let mut node = anchor(root.path(), name).await;
+        node.stop_worker();
+        let mut file = configured_file(&node);
+        file.fabric = Some(fabric_shim(root.path(), name));
+        file.transports = vec!["loopback".into(), "tailscale".into()];
+        file.save(&node.state_dir()).unwrap();
+        node.start_worker();
+        node.wait_listening().await;
+        let mut args = vec!["fleet", "transports", "add", "fabric", "--no-service"];
+        if prefer_on_add {
+            args.push("--prefer");
+        }
+        let output = node.st_ok(&args);
+        assert!(output.contains("Restart st3 daemon and st3 replication-worker"));
+        assert_eq!(
+            configured_file(&node).transports,
+            if prefer_on_add {
+                vec!["fabric", "loopback", "tailscale"]
+            } else {
+                vec!["loopback", "tailscale", "fabric"]
+            }
+        );
+        // Re-adding without preference preserves the entire configured order.
+        let before = configured_file(&node);
+        node.st_ok(&["fleet", "transports", "add", "fabric", "--no-service"]);
+        assert_eq!(configured_file(&node).transports, before.transports);
+        let endpoints = node.st_json(&["fleet", "status"])["view"].clone();
+        // Preference is local: it must not expose again or accept a new protocol.
+        fs::write(
+            root.path().join("fabric-registry").join(format!("fail-expose-{name}-fabric-id")),
+            "refuse",
+        )
+        .unwrap();
+        let output = node.st_ok(&[
+            "fleet",
+            "transports",
+            "add",
+            "fabric",
+            "--prefer",
+            "--fabric-protocol",
+            "ignored",
+            "--no-service",
+        ]);
+        if !prefer_on_add {
+            assert!(output.contains("Restart st3 daemon and st3 replication-worker"));
+        }
+        let preferred = configured_file(&node);
+        assert_eq!(preferred.transports, ["fabric", "loopback", "tailscale"]);
+        assert_eq!(preferred.fabric_protocol, before.fabric_protocol);
+        assert_eq!(node.st_json(&["fleet", "status"])["view"], endpoints);
+        let output = node.st_ok(&["fleet", "transports", "remove", "fabric", "--no-service"]);
+        assert!(output.contains("Restart st3 daemon and st3 replication-worker"));
+        assert_eq!(configured_file(&node).transports, ["loopback", "tailscale"]);
+    }
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn fabric_transport_change_refusals_leave_settings_and_endpoints_unchanged() {
     if fractal_daemon::test_support::supervise_test() {