Skip to content

feat: avoid using Option for supporters #3966

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
gilcu3 wants to merge 1 commit into from
Closed

feat: avoid using Option for supporters #3966

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion crates/node/src/coordinator.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -370,7 +370,7 @@ where
keyshare_storage: Arc<RwLock<KeyshareStorage>>,
running_state: ContractRunningState,
chain_txn_sender: TransactionSender,
foreign_chain_supporters_receiver: watch::Receiver<Option<ForeignChainSupporters>>,
foreign_chain_supporters_receiver: watch::Receiver<ForeignChainSupporters>,
block_update_receiver: tokio::sync::OwnedMutexGuard<
mpsc::UnboundedReceiver<ChainBlockUpdate>,
>,
Expand Down
128 changes: 48 additions & 80 deletions crates/node/src/foreign_chain_policy.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,20 +15,13 @@ use crate::tracking::{self, AutoAbortTask};
pub(crate) type SupportersByForeignChain = BTreeMap<dtos::ForeignChain, HashSet<ParticipantId>>;

/// Resolves the indexer's TLS-key supporters channel against the current
/// participant set, publishing each available chain's supporting participants
/// (filtered by the ForeignTx reconstruction threshold; a `None` threshold
/// means no ForeignTx domain, so no chain is available). The published value
/// is `None` until the upstream delivers its first snapshot. Must be called
/// from a tracked task; dropping the returned [`AutoAbortTask`] stops the
/// resolver, as does dropping the upstream sender or every receiver.
/// participant set. The upstream always holds a real value, so the returned
/// receiver does too. Must be called from a tracked task.
pub(crate) fn spawn_supporters_by_foreign_chain(
mut upstream: watch::Receiver<Option<ForeignChainSupporters>>,
mut upstream: watch::Receiver<ForeignChainSupporters>,
participants_config: ParticipantsConfig,
foreign_tx_reconstruction_threshold: Option<u64>,
) -> (
watch::Receiver<Option<SupportersByForeignChain>>,
AutoAbortTask<()>,
) {
) -> (watch::Receiver<SupportersByForeignChain>, AutoAbortTask<()>) {
let init_value = resolve_supporters_by_foreign_chain(
&upstream.borrow_and_update(),
&participants_config,
Expand Down Expand Up @@ -69,10 +62,10 @@ pub(crate) fn spawn_supporters_by_foreign_chain(
}

async fn await_updated_supporters(
upstream: &mut watch::Receiver<Option<ForeignChainSupporters>>,
upstream: &mut watch::Receiver<ForeignChainSupporters>,
participants_config: &ParticipantsConfig,
foreign_tx_reconstruction_threshold: Option<u64>,
) -> anyhow::Result<Option<SupportersByForeignChain>> {
) -> anyhow::Result<SupportersByForeignChain> {
upstream
.changed()
.await
Expand Down Expand Up @@ -100,27 +93,23 @@ pub(crate) fn foreign_tx_reconstruction_threshold(domains: &[dtos::DomainConfig]
/// registrations (prospective or stale ones included) and may come from a
/// different block than `participants_config`. Only supporters resolving to
/// `participants_config` — the participants this node can sign with — count
/// towards the quorum.
/// Returns `None` while no upstream snapshot exists yet: "not known" must stay
/// distinguishable from "no chain available".
/// towards the quorum. An empty map means no chain is available (either no
/// ForeignTx domain, or no chain reaches the quorum).
fn resolve_supporters_by_foreign_chain(
supporters_by_tls_key: &Option<ForeignChainSupporters>,
supporters_by_tls_key: &ForeignChainSupporters,
participants_config: &ParticipantsConfig,
foreign_tx_reconstruction_threshold: Option<u64>,
) -> Option<SupportersByForeignChain> {
let supporters_by_tls_key = supporters_by_tls_key.as_ref()?;
) -> SupportersByForeignChain {
let Some(threshold) = foreign_tx_reconstruction_threshold else {
return Some(SupportersByForeignChain::new());
return SupportersByForeignChain::new();
};
Some(
supporters_by_tls_key
.iter()
.filter_map(|(chain, tls_keys)| {
let ids = resolve_participant_ids(tls_keys, participants_config);
(ids.len() as u64 >= threshold).then_some((*chain, ids))
})
.collect(),
)
supporters_by_tls_key
.iter()
.filter_map(|(chain, tls_keys)| {
let ids = resolve_participant_ids(tls_keys, participants_config);
(ids.len() as u64 >= threshold).then_some((*chain, ids))
})
.collect()
}

/// Resolves TLS keys to the matching participants' ids; keys not belonging to a
Expand Down Expand Up @@ -198,10 +187,10 @@ mod tests {
make_participant_info(3, &keys[2]),
],
);
let supporters_by_tls_key = Some(BTreeMap::from([(
let supporters_by_tls_key = BTreeMap::from([(
dtos::ForeignChain::Bitcoin,
keys.iter().map(tls_key_for).collect::<BTreeSet<_>>(),
)]));
)]);

// When
let supporters = resolve_supporters_by_foreign_chain(
Expand All @@ -213,14 +202,14 @@ mod tests {
// Then
assert_eq!(
supporters,
Some(BTreeMap::from([(
BTreeMap::from([(
dtos::ForeignChain::Bitcoin,
HashSet::from([
ParticipantId::from_raw(1),
ParticipantId::from_raw(2),
ParticipantId::from_raw(3),
]),
)]))
)])
);
}

Expand All @@ -236,10 +225,10 @@ mod tests {
make_participant_info(2, &key2),
],
);
let supporters_by_tls_key = Some(BTreeMap::from([(
let supporters_by_tls_key = BTreeMap::from([(
dtos::ForeignChain::Bitcoin,
BTreeSet::from([tls_key_for(&key1)]),
)]));
)]);

// When
let supporters = resolve_supporters_by_foreign_chain(
Expand All @@ -249,18 +238,18 @@ mod tests {
);

// Then
assert_eq!(supporters, Some(SupportersByForeignChain::new()));
assert_eq!(supporters, SupportersByForeignChain::new());
}

#[test]
fn resolve_supporters_by_foreign_chain__should_ignore_participants_threshold() {
// Given: a participants threshold far above the single supporter.
let key1 = make_signing_key(1);
let participants_config = participants(100, vec![make_participant_info(1, &key1)]);
let supporters_by_tls_key = Some(BTreeMap::from([(
let supporters_by_tls_key = BTreeMap::from([(
dtos::ForeignChain::Bitcoin,
BTreeSet::from([tls_key_for(&key1)]),
)]));
)]);

// When
let supporters = resolve_supporters_by_foreign_chain(
Expand All @@ -270,42 +259,25 @@ mod tests {
);

// Then: only the ForeignTx domain threshold applies.
assert!(
supporters
.expect("snapshot present")
.contains_key(&dtos::ForeignChain::Bitcoin)
);
assert!(supporters.contains_key(&dtos::ForeignChain::Bitcoin));
}

#[test]
fn resolve_supporters_by_foreign_chain__should_return_empty_map_when_no_foreign_tx_domain() {
// Given: a supported chain but no ForeignTx domain (no threshold).
let key1 = make_signing_key(1);
let participants_config = participants(1, vec![make_participant_info(1, &key1)]);
let supporters_by_tls_key = Some(BTreeMap::from([(
let supporters_by_tls_key = BTreeMap::from([(
dtos::ForeignChain::Bitcoin,
BTreeSet::from([tls_key_for(&key1)]),
)]));
)]);

// When
let supporters =
resolve_supporters_by_foreign_chain(&supporters_by_tls_key, &participants_config, None);

// Then
assert_eq!(supporters, Some(SupportersByForeignChain::new()));
}

#[test]
fn resolve_supporters_by_foreign_chain__should_return_none_before_first_upstream_snapshot() {
// Given
let key1 = make_signing_key(1);
let participants_config = participants(1, vec![make_participant_info(1, &key1)]);

// When
let supporters = resolve_supporters_by_foreign_chain(&None, &participants_config, Some(1));

// Then
assert_eq!(supporters, None);
assert_eq!(supporters, SupportersByForeignChain::new());
}

#[test]
Expand Down Expand Up @@ -348,36 +320,32 @@ mod tests {
}

#[tokio::test]
async fn spawn_supporters_by_foreign_chain__should_publish_resolved_map_on_upstream_change() {
async fn spawn_supporters_by_foreign_chain__should_republish_on_upstream_change() {
let (root, _root_handle) = start_root_task("test-root", async move {
// Given: the upstream has not delivered a snapshot yet.
// Given: Bitcoin resolved as available from the first snapshot.
let key1 = make_signing_key(1);
let participants_config = participants(1, vec![make_participant_info(1, &key1)]);
let (upstream_sender, upstream_receiver) = watch::channel(None);
let (upstream_sender, upstream_receiver) = watch::channel(BTreeMap::from([(
dtos::ForeignChain::Bitcoin,
BTreeSet::from([tls_key_for(&key1)]),
)]));
let (mut supporters, _resolver_task) =
spawn_supporters_by_foreign_chain(upstream_receiver, participants_config, Some(1));
assert!(supporters.borrow().is_none());

// When: Bitcoin becomes available with the participant registered for it.
upstream_sender
.send(Some(BTreeMap::from([(
dtos::ForeignChain::Bitcoin,
BTreeSet::from([tls_key_for(&key1)]),
)])))
.unwrap();
assert!(
supporters
.borrow()
.contains_key(&dtos::ForeignChain::Bitcoin)
);

// When: the chain loses its registration upstream.
upstream_sender.send(BTreeMap::new()).unwrap();
tokio::time::timeout(Duration::from_secs(5), supporters.changed())
.await
.unwrap()
.unwrap();

// Then
assert_eq!(
*supporters.borrow(),
Some(BTreeMap::from([(
dtos::ForeignChain::Bitcoin,
HashSet::from([ParticipantId::from_raw(1)]),
)]))
);
assert_eq!(*supporters.borrow(), SupportersByForeignChain::new());
});
root.await;
}
Expand All @@ -402,12 +370,12 @@ mod tests {
)]);

// When
let (_upstream_sender, upstream_receiver) = watch::channel(Some(upstream));
let (_upstream_sender, upstream_receiver) = watch::channel(upstream);
let (supporters, _resolver_task) =
spawn_supporters_by_foreign_chain(upstream_receiver, participants_config, Some(2));

// Then: the stranger's key does not count towards the quorum.
assert_eq!(*supporters.borrow(), Some(SupportersByForeignChain::new()));
assert_eq!(*supporters.borrow(), SupportersByForeignChain::new());
});
root.await;
}
Expand Down
7 changes: 3 additions & 4 deletions crates/node/src/indexer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -584,10 +584,9 @@ pub struct IndexerAPI<TransactionSender> {
pub my_migration_info_receiver: watch::Receiver<MigrationInfo>,

/// Watcher that tracks the contract's available foreign chains and their
/// registered supporters (by TLS key). Holds `None` until the first
/// successful read after the indexer syncs.
pub foreign_chain_supporters_receiver:
watch::Receiver<Option<foreign_chain::ForeignChainSupporters>>,
/// registered supporters (by TLS key). Seeded with the first successful read
/// before the indexer hands it back, so it always holds a real value.
pub foreign_chain_supporters_receiver: watch::Receiver<foreign_chain::ForeignChainSupporters>,

pub(crate) attestation_reader: std::sync::Arc<dyn ReadAttestationExpiry>,
}
Expand Down
14 changes: 6 additions & 8 deletions crates/node/src/indexer/fake.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -545,7 +545,7 @@ struct FakeIndexerCore {
/// Broadcasts the contract state to each node.
migration_change_sender: broadcast::Sender<ContractMigrationInfo>,
/// Mirrors the real indexer's foreign-chain supporters watch channel.
foreign_chain_supporters_sender: watch::Sender<Option<ForeignChainSupporters>>,
foreign_chain_supporters_sender: watch::Sender<ForeignChainSupporters>,

/// When the core receives signature response txns, it processes them by sending them through
/// this sender. The receiver end of this is in FakeIndexManager to be received by the test
Expand Down Expand Up @@ -599,10 +599,10 @@ impl FakeIndexerCore {
state.foreign_chains_configs(),
);
foreign_chain_supporters_sender.send_if_modified(|previous| {
if previous.as_ref() == Some(&supporters) {
if *previous == supporters {
false
} else {
*previous = Some(supporters);
*previous = supporters;
true
}
});
Expand Down Expand Up @@ -881,7 +881,7 @@ pub struct FakeIndexerManager {

/// Cloned into each node's `IndexerAPI`; tracks the fake contract's
/// foreign-chain supporters.
foreign_chain_supporters_receiver: watch::Receiver<Option<ForeignChainSupporters>>,
foreign_chain_supporters_receiver: watch::Receiver<ForeignChainSupporters>,

account_id_by_uid: Arc<std::sync::Mutex<HashMap<TestNodeUid, AccountId>>>,
}
Expand Down Expand Up @@ -1071,7 +1071,7 @@ impl FakeIndexerManager {
let (verify_foreign_tx_response_sender, verify_foreign_tx_response_receiver) =
mpsc::unbounded_channel();
let (foreign_chain_supporters_sender, foreign_chain_supporters_receiver) =
watch::channel(None);
watch::channel(ForeignChainSupporters::new());
let contract = Arc::new(tokio::sync::Mutex::new(FakeMpcContractState::new()));
let account_id_by_uid = Arc::new(std::sync::Mutex::new(HashMap::new()));
let core = FakeIndexerCore {
Expand Down Expand Up @@ -1134,9 +1134,7 @@ impl FakeIndexerManager {
}

/// The supporters channel every node's `IndexerAPI` receives.
pub fn subscribe_foreign_chain_supporters(
&self,
) -> watch::Receiver<Option<ForeignChainSupporters>> {
pub fn subscribe_foreign_chain_supporters(&self) -> watch::Receiver<ForeignChainSupporters> {
self.foreign_chain_supporters_receiver.clone()
}

Expand Down
53 changes: 32 additions & 21 deletions crates/node/src/indexer/foreign_chain.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,35 +12,46 @@ const FOREIGN_CHAIN_SUPPORTERS_REFRESH_INTERVAL: Duration = Duration::from_secs(
/// TLS keys of the nodes whose registered config supports each available chain.
pub type ForeignChainSupporters = BTreeMap<dtos::ForeignChain, BTreeSet<dtos::Ed25519PublicKey>>;

/// Updates the contract's available chains mapped to their registered
/// supporters in watch channel.
/// The channel holds `None` until the first successful read; afterwards the
/// previously published value stays in effect until viewing new state from
/// contract succeeds.
/// Returns once the first supporters snapshot is read, then keeps it updated in
/// the background. Mirrors `monitor_contract_state`: the receiver always holds a
/// real value, and a failed refresh keeps the previous one.
pub async fn monitor_foreign_chain_supporters(
sender: watch::Sender<Option<ForeignChainSupporters>>,
indexer_state: Arc<IndexerState>,
) {
) -> watch::Receiver<ForeignChainSupporters> {
indexer_state.client.wait_for_full_sync().await;

loop {
let initial = loop {
match read_supporters(&indexer_state).await {
Ok(supporters) => {
sender.send_if_modified(|previous| {
if previous.as_ref() == Some(&supporters) {
false
} else {
*previous = Some(supporters);
true
}
});
}
Ok(supporters) => break supporters,
Err(e) => {
tracing::error!(target: "mpc", "error reading foreign-chain supporters from chain: {:?}", e)
tracing::error!(target: "mpc", "error reading foreign-chain supporters from chain: {:?}", e);
tokio::time::sleep(FOREIGN_CHAIN_SUPPORTERS_REFRESH_INTERVAL).await;
}
}
tokio::time::sleep(FOREIGN_CHAIN_SUPPORTERS_REFRESH_INTERVAL).await;
}
};

let (sender, receiver) = watch::channel(initial);
tokio::spawn(async move {
loop {
tokio::time::sleep(FOREIGN_CHAIN_SUPPORTERS_REFRESH_INTERVAL).await;
match read_supporters(&indexer_state).await {
Ok(supporters) => {
sender.send_if_modified(|previous| {
if *previous == supporters {
false
} else {
*previous = supporters;
true
}
});
}
Err(e) => {
tracing::error!(target: "mpc", "error reading foreign-chain supporters from chain: {:?}", e)
}
}
}
});
receiver
}

/// The two view calls are not atomic: a change finalized between them yields
Expand Down
Loading
Loading