tProxy: initialize_upstream marks successful upstream as tried_or_flagged, blocking fallback reconnect

Start in miner-apps/translator/src/lib/mod.rs around initialize_upstream, especially the checks near lines 439 and 483, then review integration-tests/tests/translator_initialize_upstream_flag_regression.rs. Run the regression test against the real pool setup and confirm that successful initialization leaves the upstream available for a later fallback retry.

Proof of Concept

diff --git a/integration-tests/tests/translator_initialize_upstream_flag_regression.rs b/integration-tests/tests/translator_initialize_upstream_flag_regression.rs
new file mode 100644
--- /dev/null
+++ b/integration-tests/tests/translator_initialize_upstream_flag_regression.rs
@@ -0,0 +1,118 @@
+use async_channel::unbounded;
+use integration_tests_sv2::{template_provider::DifficultyLevel, utils::get_available_address, *};
+use std::{convert::TryFrom, sync::Arc};
+use stratum_apps::{
+    fallback_coordinator::FallbackCoordinator,
+    key_utils::Secp256k1PublicKey,
+    task_manager::TaskManager,
+};
+use tokio_util::sync::CancellationToken;
+use translator_sv2::{
+    config::{DownstreamDifficultyConfig, TranslatorConfig, Upstream as ConfigUpstream},
+    sv1::Sv1Server,
+    utils::{TproxyMode, UpstreamEntry},
+    TranslatorSv2,
+};
+
+// Regression test for the bug at miner-apps/translator/src/lib/mod.rs:483
+// where `initialize_upstream` sets `upstream_entry.tried_or_flagged = true`
+// on the SUCCESS path. The check at line 439 then skips this entry on any
+// subsequent call (the fallback retry path), causing the translator to
+// permanently shut down after the first fallback when only one upstream
+// is configured (or to burn one upstream per fallback otherwise).
+#[tokio::test]
+async fn initialize_upstream_does_not_flag_upstream_on_success() {
+    start_tracing();
+    let (_tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low);
+    let (_pool, pool_addr, _) = start_pool(sv2_tp_config(tp_addr), vec![], vec![], false).await;
+
+    let authority_pubkey = Secp256k1PublicKey::try_from(
+        "9auqWEzQDVyd2oe1JVGFLMLHZtCo2FFqZwtKA5gd9xbuEu7PH72".to_string(),
+    )
+    .unwrap();
+
+    let listening_address = get_available_address();
+
+    let upstream_cfg = ConfigUpstream::new(
+        pool_addr.ip().to_string(),
+        pool_addr.port(),
+        authority_pubkey,
+    );
+
+    let downstream_difficulty_config = DownstreamDifficultyConfig::new(200.0, 5.0, true, 60);
+
+    let config = TranslatorConfig::new(
+        vec![upstream_cfg],
+        listening_address.ip().to_string(),
+        listening_address.port(),
+        downstream_difficulty_config,
+        2,
+        2,
+        4,
+        "regression-test".to_string(),
+        false,
+        false,
+        vec![],
+        vec![],
+        None,
+        None,
+    );
+
+    let translator = TranslatorSv2::new(config.clone());
+
+    let mut upstreams = vec![UpstreamEntry {
+        host: pool_addr.ip().to_string(),
+        port: pool_addr.port(),
+        authority_pubkey,
+        tried_or_flagged: false,
+    }];
+
+    let (_cm_to_up_sender, cm_to_up_receiver) = unbounded();
+    let (up_to_cm_sender, _up_to_cm_receiver) = unbounded();
+    let (_cm_to_sv1_sender, cm_to_sv1_receiver) = unbounded();
+    let (sv1_to_cm_sender, _sv1_to_cm_receiver) = unbounded();
+
+    let sv1_server = Arc::new(Sv1Server::new(
+        listening_address,
+        cm_to_sv1_receiver,
+        sv1_to_cm_sender,
+        config.clone(),
+        TproxyMode::NonAggregated,
+    ));
+
+    let cancellation_token = CancellationToken::new();
+    let fallback_coordinator = FallbackCoordinator::new();
+    let task_manager = Arc::new(TaskManager::new());
+
+    let result = translator
+        .initialize_upstream(
+            &mut upstreams,
+            cm_to_up_receiver,
+            up_to_cm_sender,
+            cancellation_token.clone(),
+            fallback_coordinator,
+            task_manager.clone(),
+            sv1_server,
+            vec![],
+        )
+        .await;
+
+    assert!(
+        result.is_ok(),
+        "initialize_upstream should succeed against the real pool: {result:?}"
+    );
+
+    // Regression assertion: after a successful initialization, the chosen
+    // upstream entry must NOT be flagged. Marking it (as line 483 does on
+    // HEAD) causes the translator to skip this upstream on subsequent
+    // fallback retries, leading to permanent shutdown when only one
+    // upstream is configured.
+    assert!(
+        !upstreams[0].tried_or_flagged,
+        "Upstream entry should not be marked as tried_or_flagged after successful initialization"
+    );
+
+    cancellation_token.cancel();
+    let _ =
+        tokio::time::timeout(std::time::Duration::from_secs(5), task_manager.join_all()).await;
+}