fix(topology): published party proposals only at submit

crates/decman/src/server/mod.rs

@@ -1105,6 +1105,29 @@ pub async fn start_server(
         external_parties: Default::default(),
     });
 
+    // A coordinator killed while it signed a party proposal leaves that
+    // proposal's temporary topology store on the participant. Sweep them
+    // before any run resumes, because a resumed run creates its own and a
+    // sweep running alongside would drop it (#448). Bounded, so a participant
+    // that is down does not hold up the boot.
+    match tokio::time::timeout(
+        Duration::from_secs(15),
+        workflow::proposal_store::drop_leftover_stores(&config),
+    )
+    .await
+    {
+        Ok(Ok(0)) => {}
+        Ok(Ok(dropped)) => tracing::info!(
+            dropped,
+            "Dropped temporary topology stores an earlier process left behind"
+        ),
+        Ok(Err(e)) => tracing::warn!(
+            error = format!("{e:#}"),
+            "Could not sweep leftover temporary topology stores"
+        ),
+        Err(_) => tracing::warn!("Timed out sweeping leftover temporary topology stores"),
+    }
+
     // Boot-time workflow recovery. For any `workflow_runs` row that was
     // InProgress when we shut down, re-spawn the coordinator task (which
     // resumes at the persisted `current_step` via `WorkflowState::from_persisted`)

crates/decman/src/workflow/add_party/steps/clear_onboarding.rs

@@ -195,6 +195,7 @@ pub async fn submit_clear_proposal(
         .add_transactions(tonic::Request::new(add_transactions_request(
             &synchronizer_id,
             transaction,
+            vec![],
         )))
         .await?;
 

crates/decman/src/workflow/add_party/steps/proposals/create.rs

@@ -7,7 +7,10 @@ use canton_proto_rs::com::digitalasset::canton::{
         party_to_participant::{HostingParticipant, hosting_participant},
         topology_mapping,
     },
-    topology::admin::v30::{AuthorizeRequest, ForceFlag, authorize_request},
+    topology::admin::v30::{
+        AuthorizeRequest, ForceFlag, authorize_request,
+        topology_manager_write_service_client::TopologyManagerWriteServiceClient,
+    },
 };
 use sqlx::SqlitePool;
 
@@ -19,6 +22,7 @@ use crate::{
     workflow::{
         add_party::AddPartyConfig,
         onboarding::steps::proposals::create::decode_keys_payload,
+        proposal_store,
         signing_keys::{adopt_legacy_signing_keys, own_namespace_key},
         storage::{WorkflowStorage, artifact_kinds, identity_kinds},
         topology,
@@ -92,7 +96,11 @@ pub async fn create_proposals(
 
     let synchronizer_id = utils::get_synchronizer_id(config).await?;
     let party_id = &add_party_config.decentralized_party_id;
-    let current_p2p = topology::fetch_p2p_mapping(config, &synchronizer_id, party_id).await?;
+    // The serials the proposals are pinned to, read with the mappings they
+    // are built on so the two cannot disagree.
+    let head = proposal_store::fetch_party_head(config, &synchronizer_id, party_id).await?;
+    head.ensure_namespace_is(&current_namespace_def)?;
+    let current_p2p = head.p2p.clone();
 
     let current_members: Vec<String> = current_p2p
         .participants
@@ -218,43 +226,58 @@ pub async fn create_proposals(
         }),
     };
 
-    let dns_transaction = if rehost {
-        topology::fetch_signed_namespace_definition(
-            config,
-            &synchronizer_id,
-            &new_namespace_def.decentralized_namespace,
-        )
-        .await?
-    } else {
-        tracing::info!("Creating DNS add-party proposal...");
-        topology::authorize_with_topology_retry(
-            config,
-            proposal_request(
-                &synchronizer_id,
-                topology_mapping::Mapping::DecentralizedNamespaceDefinition(
-                    new_namespace_def.clone(),
-                ),
-            ),
-            "add-party DNS",
-        )
-        .await?
-        .transaction
-        .ok_or_else(|| anyhow::anyhow!("No DNS transaction returned"))?
-    };
+    // Publishing the P2P proposal before submit (below) is safe only while it
+    // adds the new member as a host, so check that before anything is signed.
+    if !adds_host(&head.p2p, &new_p2p, &new_member_id) {
+        anyhow::bail!(
+            "The add-party P2P proposal does not add {new_member_id} as a host, so this node's \
+             signature alone could put it in force"
+        );
+    }
 
-    tracing::info!("Creating P2P add-party proposal...");
-    let p2p_response = topology::authorize_with_topology_retry(
+    // Signed without being published, so a member that never signs leaves the
+    // party unchanged (#448). A re-host leaves the namespace as it is, so no
+    // DNS is signed for it.
+    tracing::info!("Signing add-party proposals...");
+    let signed = proposal_store::sign_party_proposals(
         config,
-        proposal_request(
-            &synchronizer_id,
-            topology_mapping::Mapping::PartyToParticipant(new_p2p),
-        ),
-        "add-party P2P",
+        &synchronizer_id,
+        instance_name,
+        party_id,
+        &head,
+        (!rehost).then(|| new_namespace_def.clone()),
+        new_p2p,
+        topology::party_proposal_force_flags(),
     )
     .await?;
-    let p2p_transaction = p2p_response
-        .transaction
-        .ok_or_else(|| anyhow::anyhow!("No P2P transaction returned"))?;
+    let dns_transaction = match signed.dns {
+        Some(dns) => dns,
+        None => {
+            topology::fetch_signed_namespace_definition(
+                config,
+                &synchronizer_id,
+                &new_namespace_def.decentralized_namespace,
+            )
+            .await?
+        }
+    };
+    let p2p_transaction = signed.p2p;
+
+    // Published before submit on purpose. The new member disconnects before
+    // the mapping that hosts it is authorized (#469), and its ACS import needs
+    // that mapping in its own synchronizer store, which holds only what was
+    // sequenced. Adding a host needs that host's own signature, so the
+    // proposal stays pending until the new member signs. The DNS stays
+    // unpublished until submit, unless the signing fell back to the
+    // synchronizer above threshold 1; Canton then ignores this repeat.
+    tracing::info!("Publishing the add-party P2P proposal for the new member's store...");
+    TopologyManagerWriteServiceClient::new(config.admin_channel().await?)
+        .add_transactions(tonic::Request::new(topology::add_transactions_request(
+            &synchronizer_id,
+            p2p_transaction.clone(),
+            topology::party_proposal_force_flags(),
+        )))
+        .await?;
 
     storage
         .write_artifact(
@@ -390,6 +413,25 @@ async fn is_former_host(
     Ok(true)
 }
 
+/// Whether `proposed` hosts `new_member` and `current` does not.
+///
+/// Canton requires an added host's own signature on the mapping that adds it.
+/// A proposal that adds `new_member` therefore cannot take effect before
+/// `new_member` signs, whatever the namespace threshold.
+fn adds_host(
+    current: &PartyToParticipant,
+    proposed: &PartyToParticipant,
+    new_member: &str,
+) -> bool {
+    let hosts = |mapping: &PartyToParticipant| {
+        mapping
+            .participants
+            .iter()
+            .any(|host| host.participant_uid == new_member)
+    };
+    hosts(proposed) && !hosts(current)
+}
+
 /// The namespace fingerprint this node recorded for each member, from the
 /// identity rows written by the run that added it.
 ///
@@ -546,6 +588,42 @@ mod tests {
         assert!(format!("{err}").contains(needle), "{err}");
     }
 
+    fn hosting(uids: &[&str]) -> PartyToParticipant {
+        PartyToParticipant {
+            participants: uids
+                .iter()
+                .map(|uid| HostingParticipant {
+                    participant_uid: (*uid).to_string(),
+                    ..Default::default()
+                })
+                .collect(),
+            ..Default::default()
+        }
+    }
+
+    /// The P2P proposal is published before submit only because it adds the
+    /// new member, whose own signature Canton then requires.
+    #[test]
+    fn only_a_proposal_that_adds_the_new_member_counts_as_adding_a_host() {
+        let current = hosting(&["p1::ns", "p2::ns"]);
+        assert!(adds_host(
+            &current,
+            &hosting(&["p1::ns", "p2::ns", "p3::ns"]),
+            "p3::ns"
+        ));
+        assert!(!adds_host(&current, &current, "p3::ns"));
+        assert!(!adds_host(
+            &hosting(&["p1::ns", "p3::ns"]),
+            &hosting(&["p1::ns", "p3::ns"]),
+            "p3::ns"
+        ));
+        assert!(!adds_host(
+            &current,
+            &hosting(&["p1::ns", "p2::ns", "p4::ns"]),
+            "p3::ns"
+        ));
+    }
+
     #[sqlx::test(migrator = "MIGRATOR")]
     async fn a_namespace_that_is_not_an_owner_is_a_plain_add(pool: SqlitePool) -> Result {
         let fingerprint = utils::compute_fingerprint(&key(1, SigningKeyUsage::Namespace));

crates/decman/src/workflow/add_party/steps/proposals/submit.rs

@@ -84,6 +84,7 @@ pub async fn submit_proposals(
             &synchronizer_id,
             "add-party",
             p2p_transaction,
+            topology::party_proposal_force_flags(),
             confirm_p2p,
         )
         .await?;
@@ -103,8 +104,11 @@ pub async fn submit_proposals(
             config,
             &synchronizer_id,
             "add-party",
-            dns_transaction,
-            p2p_transaction,
+            topology::DnsP2pSubmission {
+                dns: dns_transaction,
+                p2p: p2p_transaction,
+                force_changes: topology::party_proposal_force_flags(),
+            },
             || {
                 wait_for_owners(
                     config,

crates/decman/src/workflow/change_threshold/steps/proposals/create.rs

@@ -1,13 +1,6 @@
 use canton_proto_rs::com::digitalasset::canton::{
     crypto::v30::SigningKeysWithThreshold,
-    protocol::v30::{
-        DecentralizedNamespaceDefinition, PartyToParticipant, TopologyMapping, enums,
-        topology_mapping,
-    },
-    topology::admin::v30::{
-        AuthorizeRequest, authorize_request,
-        topology_manager_write_service_client::TopologyManagerWriteServiceClient,
-    },
+    protocol::v30::{DecentralizedNamespaceDefinition, PartyToParticipant},
 };
 use sqlx::SqlitePool;
 
@@ -18,6 +11,7 @@ use crate::{
     utils,
     workflow::{
         change_threshold::ChangeThresholdConfig,
+        proposal_store,
         signing_keys::adopt_legacy_signing_keys,
         storage::{WorkflowStorage, artifact_kinds},
         topology,
@@ -91,7 +85,11 @@ pub async fn create_proposals(
     let synchronizer_id = utils::get_synchronizer_id(config).await?;
     tracing::debug!("Using synchronizer ID: {synchronizer_id}");
 
-    let current_p2p = topology::fetch_p2p_mapping(config, &synchronizer_id, &party_id).await?;
+    // The serials the proposals are pinned to, read with the mappings they
+    // are built on so the two cannot disagree.
+    let head = proposal_store::fetch_party_head(config, &synchronizer_id, &party_id).await?;
+    head.ensure_namespace_is(&current_namespace_def)?;
+    let current_p2p = head.p2p.clone();
     tracing::info!(
         "Current P2P mapping has {count} participant(s), threshold {threshold}",
         count = current_p2p.participants.len(),
@@ -140,58 +138,24 @@ pub async fn create_proposals(
         }),
     };
 
-    // Create proposals using topology manager
-    let mut topology_client = TopologyManagerWriteServiceClient::new(config.admin_channel().await?);
-
-    // Create DNS proposal
-    tracing::info!("Creating DNS change-threshold proposal...");
-    let dns_request = tonic::Request::new(AuthorizeRequest {
-        r#type: Some(authorize_request::Type::Proposal(
-            authorize_request::Proposal {
-                change: enums::TopologyChangeOp::AddReplace as i32,
-                serial: 0,
-                mapping: Some(authorize_request::proposal::Mapping::V30(TopologyMapping {
-                    mapping: Some(topology_mapping::Mapping::DecentralizedNamespaceDefinition(
-                        new_namespace_def.clone(),
-                    )),
-                })),
-            },
-        )),
-        must_fully_authorize: false,
-        force_changes: vec![],
-        signed_by: vec![],
-        store: Some(topology::synchronizer_store_id(&synchronizer_id)),
-        wait_to_become_effective: None,
-    });
-
-    let dns_response = topology_client.authorize(dns_request).await?.into_inner();
-    let dns_transaction = dns_response
-        .transaction
-        .ok_or_else(|| anyhow::anyhow!("No DNS transaction returned"))?;
-
-    // Create P2P proposal
-    tracing::info!("Creating P2P change-threshold proposal...");
-    let p2p_request = tonic::Request::new(AuthorizeRequest {
-        r#type: Some(authorize_request::Type::Proposal(
-            authorize_request::Proposal {
-                change: enums::TopologyChangeOp::AddReplace as i32,
-                serial: 0,
-                mapping: Some(authorize_request::proposal::Mapping::V30(TopologyMapping {
-                    mapping: Some(topology_mapping::Mapping::PartyToParticipant(new_p2p)),
-                })),
-            },
-        )),
-        must_fully_authorize: false,
-        force_changes: vec![],
-        signed_by: vec![],
-        store: Some(topology::synchronizer_store_id(&synchronizer_id)),
-        wait_to_become_effective: None,
-    });
-
-    let p2p_response = topology_client.authorize(p2p_request).await?.into_inner();
-    let p2p_transaction = p2p_response
-        .transaction
-        .ok_or_else(|| anyhow::anyhow!("No P2P transaction returned"))?;
+    // Signed without being published: nothing reaches the synchronizer until
+    // submit, so a member that never signs leaves the party unchanged (#448).
+    tracing::info!("Signing change-threshold proposals...");
+    let signed = proposal_store::sign_party_proposals(
+        config,
+        &synchronizer_id,
+        instance_name,
+        &party_id,
+        &head,
+        Some(new_namespace_def.clone()),
+        new_p2p,
+        topology::party_proposal_force_flags(),
+    )
+    .await?;
+    let dns_transaction = signed
+        .dns
+        .ok_or_else(|| anyhow::anyhow!("No DNS change-threshold proposal was signed"))?;
+    let p2p_transaction = signed.p2p;
 
     // Persist proposals + supporting data. Each protobuf is written with the
     // same `varint(len)||proto` framing the file path used, so

crates/decman/src/workflow/change_threshold/steps/proposals/submit.rs

@@ -96,8 +96,11 @@ pub async fn submit_change(
         config,
         &synchronizer_id,
         "change-threshold",
-        dns_transaction,
-        p2p_transaction,
+        topology::DnsP2pSubmission {
+            dns: dns_transaction,
+            p2p: p2p_transaction,
+            force_changes: topology::party_proposal_force_flags(),
+        },
         || {
             wait_for_dns_in_topology(
                 config,