fix(node): keep attestation tasks alive after submission retry timeout

[pbeza]

Closes #2287

[pbeza]

A few points worth highlighting for reviewers before they start reviewing:

1. I did a small refactor and introduced a shared generate_and_submit function used by both periodic_attestation_submission and monitor_attestation_removal. There was quite a bit of copy-paste between them:
   

   mpc/crates/node/src/tee/remote_attestation.rs

   Lines 298 to 356 in 9ce222e

   	let fresh_attestation = match tee_authority
   	.generate_attestation(report_data.clone())
   	.await
   	{
   	Ok(att) => {
   	crate::metrics::MPC_TEE_ATTESTATION_ATTEMPTS_TOTAL
   	.with_label_values(&[crate::metrics::MPC_TEE_ATTESTATION_OUTCOME_SUCCESS])
   	.inc();
   	att
   	}
   	Err(tee_authority::tee_authority::AttestationError::CollateralFetch(e)) => {
   	crate::metrics::MPC_TEE_ATTESTATION_ATTEMPTS_TOTAL
   	.with_label_values(&[crate::metrics::MPC_TEE_ATTESTATION_OUTCOME_FAILURE])
   	.inc();
   	tracing::warn!(
   	error = %e,
   	"TEE attestation failed, periodic attestation task will retry",
   	);
   	was_available = is_available;
   	continue;
   	}
   	Err(e) => {
   	crate::metrics::MPC_TEE_ATTESTATION_ATTEMPTS_TOTAL
   	.with_label_values(&[crate::metrics::MPC_TEE_ATTESTATION_OUTCOME_FAILURE])
   	.inc();
   	return Err(
   	anyhow::anyhow!(e).context("TEE attestation failed, cannot continue")
   	);
   	}
   	};
   	let allowed_image_hashes_in_contract: Vec<_> = allowed_image_hashes_in_contract
   	.borrow()
   	.iter()
   	.map(|entry| entry.image_hash)
   	.collect();
   	let allowed_launcher_compose_hashes_in_contract =
   	allowed_launcher_compose_hashes_in_contract.borrow().clone();
   	let pre_submit_expiry = match attestation_reader
   	.read_stored_attestation_expiry(&tls_public_key)
   	.await
   	{
   	Ok(baseline) => baseline, // None just means nothing stored yet (e.g. first submit)
   	// Submit anyway on a read error: re-submitting a removed attestation is the
   	// priority, and a broken read must not block it (confirmation just lacks a baseline).
   	Err(error) => {
   	tracing::warn!(%error, "could not read pre-submit attestation baseline; submitting without it");
   	None
   	}
   	};
   	validate_and_submit_remote_attestation(
   	tx_sender.clone(),
   	fresh_attestation.clone(),
   	tls_public_key.clone(),
   	account_public_key.clone(),
   	&allowed_image_hashes_in_contract,
   	&allowed_launcher_compose_hashes_in_contract,
   	pre_submit_expiry,
   	)
   	.await?;

   mpc/crates/node/src/tee/remote_attestation.rs

   Lines 179 to 231 in 9ce222e

   	let fresh_attestation = match tee_authority
   	.generate_attestation(report_data.clone())
   	.await
   	{
   	Ok(att) => {
   	crate::metrics::MPC_TEE_ATTESTATION_ATTEMPTS_TOTAL
   	.with_label_values(&[crate::metrics::MPC_TEE_ATTESTATION_OUTCOME_SUCCESS])
   	.inc();
   	att
   	}
   	Err(tee_authority::tee_authority::AttestationError::CollateralFetch(e)) => {
   	crate::metrics::MPC_TEE_ATTESTATION_ATTEMPTS_TOTAL
   	.with_label_values(&[crate::metrics::MPC_TEE_ATTESTATION_OUTCOME_FAILURE])
   	.inc();
   	tracing::warn!(error = %e, "TEE attestation failed, will retry next interval");
   	continue;
   	}
   	Err(e) => {
   	crate::metrics::MPC_TEE_ATTESTATION_ATTEMPTS_TOTAL
   	.with_label_values(&[crate::metrics::MPC_TEE_ATTESTATION_OUTCOME_FAILURE])
   	.inc();
   	return Err(anyhow::anyhow!(e).context("TEE attestation failed, cannot continue"));
   	}
   	};
   	let allowed_image_hashes_in_contract: Vec<_> = allowed_image_hashes_in_contract
   	.borrow()
   	.iter()
   	.map(|entry| entry.image_hash)
   	.collect();
   	let allowed_launcher_compose_hashes_in_contract =
   	allowed_launcher_compose_hashes_in_contract.borrow().clone();
   	let pre_submit_expiry = match attestation_reader
   	.read_stored_attestation_expiry(&tls_public_key)
   	.await
   	{
   	Ok(baseline) => baseline, // None just means nothing stored yet (e.g. first submit)
   	// Submit anyway on a read error: refreshing the attestation is the priority, and a
   	// broken read must not block submission (the confirmation just can't use a baseline).
   	Err(error) => {
   	tracing::warn!(%error, "could not read pre-submit attestation baseline; submitting without it");
   	None
   	}
   	};
   	validate_and_submit_remote_attestation(
   	tx_sender.clone(),
   	fresh_attestation.clone(),
   	tls_public_key.clone(),
   	account_public_key.clone(),
   	&allowed_image_hashes_in_contract,
   	&allowed_launcher_compose_hashes_in_contract,
   	pre_submit_expiry,
   	)
   	.await?;

2. I added:

   mpc/crates/node/src/run.rs

   Lines 371 to 373 in 38e27a2

   	// A failed submission can retry internally for longer than the interval; skip the missed
   	// ticks instead of bursting stale submissions afterwards
   	attestation_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

   to make sure we don't burst with attestation resubmission retries (tokio::time::interval defaults to MissedTickBehavior::Burst).

3. I added a new metric:

   mpc/crates/node/src/metrics.rs

   Lines 480 to 490 in 38e27a2

   	/// A node that cannot get its attestation onto the contract retries forever rather than giving
   	/// up, so this counter is the only signal an operator has that the node is failing to attest.
   	pub static MPC_TEE_ATTESTATION_SUBMISSIONS_TOTAL: LazyLock<prometheus::IntCounterVec> =
   	LazyLock::new(|| {
   	prometheus::register_int_counter_vec!(
   	"mpc_tee_attestation_submissions_total",
   	"Total number of TEE attestation submissions to the MPC contract",
   	&["outcome"],
   	)
   	.unwrap()
   	});

   (We should probably set up an alert for that metric.)

   The existing mpc_tee_attestation_attempts_total only counts attestation generation. Since a permanently failing submission is now a quiet infinite retry loop rather than causing the task to die, operators need something they can alert on.

4. I reworked the tests: renamed to the repo's <unit>__should_<behavior> convention, and split test_tee_attestation_removal_detection into focused cases. The per-test setup (keys, watch channels, submitter) is now shared:
   

   mpc/crates/node/src/tee/remote_attestation.rs

   Lines 430 to 492 in 38e27a2

   	struct TestSetup {
   	node_id: NodeId,
   	tee_accounts_sender: watch::Sender<Vec<NodeId>>,
   	submitter: AttestationSubmitter<MockSender>,
   	}
   	/// Builds an [`AttestationSubmitter`] around a [`MockSender`], with the node initially
   	/// present in the TEE accounts watch channel.
   	fn test_setup() -> TestSetup {
   	let (tls_public_key, account_public_key) = test_keys();
   	let node_id = NodeId {
   	account_id: "test_node.near".parse().unwrap(),
   	tls_public_key: tls_public_key.clone(),
   	account_public_key: account_public_key.clone(),
   	};
   	let (tee_accounts_sender, _) = watch::channel(vec![node_id.clone()]);
   	let (_, allowed_image_hashes) = watch::channel(vec![]);
   	let (_, allowed_launcher_compose_hashes) = watch::channel(vec![]);
   	let submitter = AttestationSubmitter {
   	tee_authority: TeeAuthority::from(LocalTeeAuthorityConfig::default()),
   	tx_sender: MockSender::default(),
   	tls_public_key,
   	account_public_key,
   	allowed_image_hashes,
   	allowed_launcher_compose_hashes,
   	attestation_reader: Arc::new(StubAttestationExpiryReader { fail: false }),
   	};
   	TestSetup {
   	node_id,
   	tee_accounts_sender,
   	submitter,
   	}
   	}
   	impl TestSetup {
   	fn sender(&self) -> &MockSender {
   	&self.submitter.tx_sender
   	}
   	fn spawn_periodic(&self, ticks: usize) -> tokio::task::JoinHandle<()> {
   	tokio::spawn(periodic_attestation_submission(
   	self.submitter.clone(),
   	MockTicker::new(ticks),
   	))
   	}
   	fn spawn_monitor(&self) -> tokio::task::JoinHandle<()> {
   	tokio::spawn(monitor_attestation_removal(
   	self.submitter.clone(),
   	self.node_id.account_id.clone(),
   	self.tee_accounts_sender.subscribe(),
   	))
   	}
   	fn remove_node_from_tee_accounts(&self) {
   	self.tee_accounts_sender.send_replace(vec![]);
   	}
   	fn add_node_to_tee_accounts(&self) {
   	self.tee_accounts_sender
   	.send_replace(vec![self.node_id.clone()]);
   	}
   	}

   MockSender also gained a set_failing switch. Nothing could simulate a failed submission before.

   I dropped two assertions. They were not separate tests, they were the tail of test_tee_attestation_removal_detection: Expected submission count to remain stable after stopping monitoring service and Expected no resubmission when monitoring service is stopped. Both run after monitoring_task.abort(), so they assert that an aborted task does nothing, which is a property of JoinHandle::abort rather than of our code:
   

   mpc/crates/node/src/tee/remote_attestation.rs

   Lines 566 to 668 in 9ce222e

   	#[tokio::test]
   	async fn test_tee_attestation_removal_detection() {
   	let node_account_id: AccountId = "test_node.near".parse().unwrap();
   	let mut rng = rand::rngs::StdRng::seed_from_u64(42);
   	let tls_public_key: Ed25519PublicKey =
   	(&SigningKey::generate(&mut rng).verifying_key()).into();
   	let account_public_key: Ed25519PublicKey =
   	(&SigningKey::generate(&mut rng).verifying_key()).into();
   	let tee_authority = TeeAuthority::from(LocalTeeAuthorityConfig::default());
   	let node_id = NodeId {
   	account_id: node_account_id.clone(),
   	tls_public_key: tls_public_key.clone(),
   	account_public_key: account_public_key.clone(),
   	};
   	// Create initial TEE accounts list including our node
   	let initial_tee_accounts = vec![node_id.clone()];
   	let (tee_accounts_sender, receiver) = watch::channel(initial_tee_accounts);
   	let (_, allowed_image_hashes_receiver) = watch::channel(vec![]);
   	let (_, allowed_launcher_compose_hashes_receiver) = watch::channel(vec![]);
   	// Create mock sender with contract simulator built-in
   	let mock_sender = MockSender::new(tee_accounts_sender.clone(), node_id.clone());
   	let submitter = AttestationSubmitter {
   	tee_authority,
   	tx_sender: mock_sender.clone(),
   	tls_public_key,
   	account_public_key,
   	allowed_image_hashes: allowed_image_hashes_receiver,
   	allowed_launcher_compose_hashes: allowed_launcher_compose_hashes_receiver,
   	attestation_reader: Arc::new(StubAttestationExpiryReader),
   	};
   	let monitoring_task = tokio::spawn(monitor_attestation_removal(
   	submitter,
   	node_account_id.clone(),
   	receiver,
   	));
   	// Yield control to allow the monitoring task to start and process initial state.
   	// This is preferred over sleep() as it doesn't introduce arbitrary timing delays
   	tokio::task::yield_now().await;
   	// Verify no submission occurred initially (node is in TEE accounts)
   	assert_eq!(mock_sender.count(), 0);
   	// Remove the node from TEE accounts (simulate attestation removal)
   	let removed_tee_accounts = vec![]; // Node is no longer in TEE accounts
   	tee_accounts_sender.send(removed_tee_accounts).unwrap();
   	// Wait for the resubmission to occur (with timeout to avoid hanging)
   	tokio::time::timeout(
   	TEST_EXPECTED_ATTESTATION_RESUBMISSION_TIMEOUT,
   	mock_sender.wait_for_submission(),
   	)
   	.await
   	.expect("Expected resubmission to occur within timeout");
   	// Verify attestation resubmission occurred and no additional submissions occurred
   	// (node should be back in TEE accounts automatically after resubmission)
   	assert_eq!(
   	mock_sender.count(),
   	1,
   	"Expected exactly one resubmission when node was removed"
   	);
   	// Stop monitoring service and verify no further submissions occur
   	monitoring_task.abort();
   	let _ = monitoring_task.await;
   	// Verify the submission count remains unchanged after stopping monitoring
   	assert_eq!(
   	mock_sender.count(),
   	1,
   	"Expected submission count to remain stable after stopping monitoring service"
   	);
   	// Remove the node from TEE accounts again to verify monitoring service is truly stopped
   	let removed_tee_accounts = vec![]; // Node is no longer in TEE accounts
   	let _ = tee_accounts_sender.send(removed_tee_accounts);
   	// Give a brief moment to ensure no resubmission occurs when monitoring is stopped
   	// Since the monitoring task is stopped, we use a timeout to verify no submission happens
   	let timeout_result = tokio::time::timeout(
   	TEST_VERIFY_NO_ATTESTATION_RESUBMISSION_TIMEOUT,
   	mock_sender.wait_for_submission(),
   	)
   	.await;
   	// Verify the timeout occurred (no submission)
   	assert!(
   	timeout_result.is_err(),
   	"Expected no resubmission when monitoring service is stopped"
   	);
   	// Verify no resubmission occurred (monitoring service is stopped)
   	assert_eq!(
   	mock_sender.count(),
   	1,
   	"Expected no resubmission when monitoring service is stopped"
   	);
   	}

   They are replaced by monitor_attestation_removal__should_stop_when_watch_channel_closes (the real termination path) and monitor_attestation_removal__should_resubmit_on_each_removal. The five new behaviors: surviving the 12h submission retry timeout, retrying a failed resubmission on the fixed delay, retrying it on the next TEE-accounts update, resubmitting every time the node goes from present to absent rather than only the first time, and stopping when the watch channel closes.

Despite all these changes, the PR is still relatively small, so I think it’s fine not to split the refactor into a separate PR.
The extracted code also helped keep the new tests smaller.

[gilcu3]

Looks good for what it was trying to do. Left some minor comments, but will come back after some slack discussion, as I am not sure about the approach

[gilcu3]

same, I don't think we need to update this old doc, we could actually remove this one to reduce the noise

[pbeza]

Ditto: #4201 (comment).

[gilcu3]

Thank you! I think all my comments were already there :)

[barakeinav1]

I did only a shallow review. but LGTM