Merge remote-tracking branch 'origin/dev' into marvin/incremental-updates-1

# Conflicts:
#	test_fixtures/fixtures/prebuilt_sequencer_db.dump

integration_tests/src/utils.rs

@@ -1,10 +1,14 @@
 use std::time::Duration;
 
-use anyhow::Context as _;
+use anyhow::{Context as _, Result, ensure};
 use key_protocol::key_management::key_tree::chain_index::ChainIndex;
 use lee_core::account::AccountId;
 use log::info;
-use sequencer_service_rpc::RpcClient as _;
+use sequencer_core::{
+    block_publisher::{Ed25519PublicKey, read_channel_state},
+    config::BedrockConfig,
+};
+use sequencer_service_rpc::{RpcClient as _, SequencerClient};
 use test_fixtures::{TIME_TO_WAIT_FOR_BLOCK_SECONDS, TestContext, verify_commitment_is_in_state};
 use wallet::{
     AccountIdentity,
@@ -21,6 +25,71 @@ use wallet::{
 
 /// Maximum time to wait for the indexer to catch up to the sequencer.
 pub const L2_TO_L1_TIMEOUT: Duration = Duration::from_mins(6);
+/// Maximum time a single [`wait_until`] may poll before giving up.
+const PHASE_TIMEOUT: Duration = Duration::from_secs(360);
+const POLL_INTERVAL: Duration = Duration::from_secs(2);
+
+/// Polls `check` until it reports ready, failing with `what` on timeout.
+pub async fn wait_until<F, Fut>(what: &str, mut check: F) -> Result<()>
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<bool>>,
+{
+    let wait = async {
+        while !check().await? {
+            tokio::time::sleep(POLL_INTERVAL).await;
+        }
+        Ok::<(), anyhow::Error>(())
+    };
+    tokio::time::timeout(PHASE_TIMEOUT, wait)
+        .await
+        .with_context(|| format!("Timed out waiting for {what}"))?
+}
+
+/// The channel's accredited keys, sorted, plus whose turn the tip was written on.
+pub async fn committee(
+    config: &BedrockConfig,
+) -> Result<(Vec<[u8; 32]>, Option<Ed25519PublicKey>)> {
+    let Some(state) = read_channel_state(config).await? else {
+        return Ok((Vec::new(), None));
+    };
+    let turn = state
+        .accredited_keys
+        .get(usize::from(state.tip_sequencer))
+        .copied();
+    let mut keys: Vec<_> = state
+        .accredited_keys
+        .iter()
+        .map(Ed25519PublicKey::to_bytes)
+        .collect();
+    keys.sort_unstable();
+    Ok((keys, turn))
+}
+
+/// Asserts A and B hold byte-identical block hashes over their common prefix.
+pub async fn assert_same_chain(a: &SequencerClient, b: &SequencerClient) -> Result<()> {
+    let common = a
+        .get_last_block_id()
+        .await?
+        .min(b.get_last_block_id().await?);
+    for id in 1..=common {
+        let block_a = a
+            .get_block(id)
+            .await?
+            .with_context(|| format!("A is missing block {id}"))?;
+        let block_b = b
+            .get_block(id)
+            .await?
+            .with_context(|| format!("B is missing block {id}"))?;
+        ensure!(
+            block_a.header.hash == block_b.header.hash,
+            "Chain divergence at block {id}: A {:?} vs B {:?}",
+            block_a.header.hash,
+            block_b.header.hash
+        );
+    }
+    Ok(())
+}
 
 /// Create a private or public account at the given chain index and return its ID.
 /// Pass `cci: None` to use the wallet's next available chain index.

integration_tests/tests/multi_sequencer.rs

@@ -12,14 +12,12 @@ use std::time::Duration;
 use anyhow::{Context as _, Result, ensure};
 use indexer_service_rpc::RpcClient as _;
 use integration_tests::{
+    assert_same_chain, committee,
     config::{self, SequencerPartialConfig},
-    init_logger,
+    init_logger, wait_until,
 };
 use logos_blockchain_key_management_system_service::keys::Ed25519Key;
-use sequencer_core::{
-    block_publisher::{Ed25519PublicKey, read_channel_state},
-    config::BedrockConfig,
-};
+use sequencer_core::config::BedrockConfig;
 use sequencer_service_rpc::{RpcClient as _, SequencerClient};
 use test_fixtures::{
     MultiZoneTestContextBuilder, ZoneTestContextBuilder, config::MultiNodeTestContextConfig,
@@ -27,8 +25,6 @@ use test_fixtures::{
 use testnet_initial_state::{initial_pub_accounts_private_keys, initial_public_user_accounts};
 use tokio::test;
 
-const PHASE_TIMEOUT: Duration = Duration::from_secs(360);
-const POLL_INTERVAL: Duration = Duration::from_secs(2);
 const TRANSFER_AMOUNT: u128 = 10;
 /// ≈4 turn windows, at the `system_accounts` posting timeframe and 5 s blocks.
 const ROTATION_BLOCKS: u64 = 8;
@@ -158,23 +154,6 @@ async fn multi_sequencer_committee_converges() -> Result<()> {
     Ok(())
 }
 
-/// Polls `check` until it reports ready, failing with `what` on timeout.
-async fn wait_until<F, Fut>(what: &str, mut check: F) -> Result<()>
-where
-    F: FnMut() -> Fut,
-    Fut: Future<Output = Result<bool>>,
-{
-    let wait = async {
-        while !check().await? {
-            tokio::time::sleep(POLL_INTERVAL).await;
-        }
-        Ok::<(), anyhow::Error>(())
-    };
-    tokio::time::timeout(PHASE_TIMEOUT, wait)
-        .await
-        .with_context(|| format!("Timed out waiting for {what}"))?
-}
-
 /// Polls the sequencer until its chain height reaches `target`.
 async fn wait_for_height(client: &SequencerClient, target: u64, what: &str) -> Result<()> {
     wait_until(&format!("{what} (target height {target})"), || async {
@@ -182,46 +161,3 @@ async fn wait_for_height(client: &SequencerClient, target: u64, what: &str) -> R
     })
     .await
 }
-
-/// The channel's accredited keys, sorted, plus whose turn the tip was written on.
-async fn committee(config: &BedrockConfig) -> Result<(Vec<[u8; 32]>, Option<Ed25519PublicKey>)> {
-    let Some(state) = read_channel_state(config).await? else {
-        return Ok((Vec::new(), None));
-    };
-    let turn = state
-        .accredited_keys
-        .get(usize::from(state.tip_sequencer))
-        .copied();
-    let mut keys: Vec<_> = state
-        .accredited_keys
-        .iter()
-        .map(Ed25519PublicKey::to_bytes)
-        .collect();
-    keys.sort_unstable();
-    Ok((keys, turn))
-}
-
-/// Asserts A and B hold byte-identical block hashes over their common prefix.
-async fn assert_same_chain(a: &SequencerClient, b: &SequencerClient) -> Result<()> {
-    let common = a
-        .get_last_block_id()
-        .await?
-        .min(b.get_last_block_id().await?);
-    for id in 1..=common {
-        let block_a = a
-            .get_block(id)
-            .await?
-            .with_context(|| format!("A is missing block {id}"))?;
-        let block_b = b
-            .get_block(id)
-            .await?
-            .with_context(|| format!("B is missing block {id}"))?;
-        ensure!(
-            block_a.header.hash == block_b.header.hash,
-            "Chain divergence at block {id}: A {:?} vs B {:?}",
-            block_a.header.hash,
-            block_b.header.hash
-        );
-    }
-    Ok(())
-}

integration_tests/tests/sequencer_rejoin.rs

@@ -0,0 +1,213 @@
+#![expect(
+    clippy::tests_outside_test_module,
+    reason = "top-level test functions are conventional for integration tests"
+)]
+
+//! B unstakes in full, leaves the committee, gets its stake back, then rejoins.
+
+use std::time::Duration;
+
+use anyhow::{Context as _, Result, ensure};
+use integration_tests::{
+    account_balance, assert_same_chain, committee, get_account, init_logger, wait_until,
+};
+use lee::{AccountId, PublicKey, program::Program};
+use log::info;
+use logos_blockchain_key_management_system_service::keys::Ed25519Key;
+use sequencer_core::config::BedrockConfig;
+use sequencer_service_rpc::RpcClient as _;
+use test_fixtures::{
+    MultiZoneTestContextBuilder, TestContext, ZoneTestContextBuilder,
+    config::{self, MultiNodeTestContextConfig, SequencerPartialConfig},
+};
+use tokio::test;
+use wallet::AccountIdentity;
+
+/// What genesis stakes each founding sequencer.
+const STAKE: u128 = system_accounts::DEFAULT_MINIMUM_SEQUENCER_STAKE;
+
+#[test]
+async fn a_sequencer_leaves_the_committee_and_rejoins() -> Result<()> {
+    init_logger();
+
+    let channel = config::bedrock_channel_id();
+    let partial = SequencerPartialConfig {
+        block_create_timeout: Duration::from_secs(2),
+        ..SequencerPartialConfig::default()
+    };
+
+    let mut ctx = MultiZoneTestContextBuilder::default()
+        .with_zone(
+            ZoneTestContextBuilder::new(MultiNodeTestContextConfig {
+                num_nodes: 2,
+                bedrock_channel: channel,
+            })
+            .with_sequencer_partial_config(partial),
+        )
+        .build()
+        .await
+        .context("Failed to build the two-sequencer test context")?;
+
+    let key_b = Ed25519Key::from_bytes(&config::sequencer_signing_key_from_seed(1)).public_key();
+    let stake_key_b = sequencer_stake_core::SequencerKey::new(key_b.to_bytes())
+        .context("Sequencer B's Bedrock key is not a valid Ed25519 point")?;
+
+    let bedrock_config = BedrockConfig {
+        channel_id: channel,
+        node_url: config::addr_to_url(config::UrlProtocol::Http, ctx.bedrock_addr())?,
+        funding_key: config::bedrock_funding_key(),
+        auth: None,
+        priority_fee: sequencer_core::config::default_priority_fee(),
+    };
+
+    // B's genesis stake sits on an account only this key can sign for.
+    let owner_b = config::founding_stake_owner_key(1)?;
+    let ownership_b = AccountId::from(&PublicKey::new_from_private_key(&owner_b));
+    ctx.wallet_mut()
+        .storage_mut()
+        .key_chain_mut()
+        .add_imported_public_account(owner_b);
+
+    let config_id = system_accounts::sequencer_stake_config_account_id();
+
+    let settlement = AccountId::from(&PublicKey::new_from_private_key(
+        &config::default_public_accounts_for_wallet()[0].0,
+    ));
+
+    wait_until("both staked keys to be accredited", || async {
+        Ok(committee(&bedrock_config)
+            .await?
+            .0
+            .contains(&key_b.to_bytes()))
+    })
+    .await?;
+    info!("Both sequencers accredited from channel creation");
+
+    // B leaves.
+    let settlement_before = account_balance(&ctx, settlement).await?;
+    send_stake_tx(
+        &ctx,
+        vec![
+            AccountIdentity::Public(ownership_b),
+            AccountIdentity::PublicNoSign(config_id),
+        ],
+        &sequencer_stake_core::Instruction::UnstakeRequest {
+            amount: STAKE,
+            destination: settlement,
+        },
+    )
+    .await
+    .context("Failed to submit B's UnstakeRequest")?;
+    info!("B requested a full unstake");
+
+    wait_until("B to leave the committee", || async {
+        Ok(!committee(&bedrock_config)
+            .await?
+            .0
+            .contains(&key_b.to_bytes()))
+    })
+    .await?;
+    info!("B removed from the Bedrock committee");
+
+    wait_until("B's stake to be released", || async {
+        Ok(get_account(&ctx, ownership_b).await?.balance == 0)
+    })
+    .await?;
+    ensure!(
+        account_balance(&ctx, settlement).await? == settlement_before + STAKE,
+        "the released stake should have reached the settlement account"
+    );
+    ensure!(
+        stake_entry(&ctx, stake_key_b).await?.is_none(),
+        "a fully released key should have no config entry left"
+    );
+    info!("B's stake released in full");
+
+    // B rejoins on the same ownership account, which stays claimed after an exit.
+    let mover_instruction_data =
+        Program::serialize_instruction(authenticated_transfer_core::Instruction::Transfer {
+            amount: STAKE,
+        })
+        .context("Failed to serialize the mover instruction")?;
+    send_stake_tx(
+        &ctx,
+        vec![
+            AccountIdentity::Public(settlement),
+            AccountIdentity::Public(ownership_b),
+            AccountIdentity::PublicNoSign(config_id),
+        ],
+        &sequencer_stake_core::Instruction::Stake {
+            sequencer_key: stake_key_b,
+            amount: STAKE,
+            mover_program_id: programs::authenticated_transfer().id(),
+            mover_instruction_data,
+        },
+    )
+    .await
+    .context("Failed to submit B's re-stake")?;
+
+    wait_until("B's re-stake to land", || async {
+        Ok(get_account(&ctx, ownership_b).await?.balance == STAKE)
+    })
+    .await?;
+    info!("B staked again");
+
+    wait_until("B to be accredited again", || async {
+        Ok(committee(&bedrock_config)
+            .await?
+            .0
+            .contains(&key_b.to_bytes()))
+    })
+    .await?;
+    info!("B back in the Bedrock committee");
+
+    // Rejoining is only real if B writes to the channel again.
+    wait_until("the round-robin turn to reach B again", || async {
+        Ok(committee(&bedrock_config).await?.1 == Some(key_b))
+    })
+    .await?;
+
+    let a = ctx
+        .sequencer_client_by_node_ids(channel, 0)
+        .context("Missing sequencer A")?;
+    let b = ctx
+        .sequencer_client_by_node_ids(channel, 1)
+        .context("Missing sequencer B")?;
+    let resumed_at = a.get_last_block_id().await?;
+    wait_until("the chain to advance past B's rejoin", || async {
+        Ok(a.get_last_block_id().await? > resumed_at)
+    })
+    .await?;
+    assert_same_chain(a, b).await?;
+    info!("B produces again and both sequencers agree on the chain");
+
+    Ok(())
+}
+
+/// Sends `instruction` to `sequencer_stake` over `accounts`.
+async fn send_stake_tx(
+    ctx: &TestContext,
+    accounts: Vec<AccountIdentity>,
+    instruction: &sequencer_stake_core::Instruction,
+) -> Result<()> {
+    let data = Program::serialize_instruction(instruction.clone())
+        .context("Failed to serialize the sequencer_stake instruction")?;
+    ctx.wallet()
+        .send_pub_tx(accounts, data, programs::sequencer_stake().id())
+        .await
+        .map_err(|err| anyhow::anyhow!("Failed to submit sequencer_stake transaction: {err:?}"))?;
+    Ok(())
+}
+
+/// The `sequencer_stake` config entry for `sequencer_key`, if any.
+async fn stake_entry(
+    ctx: &TestContext,
+    sequencer_key: sequencer_stake_core::SequencerKey,
+) -> Result<Option<sequencer_stake_core::SequencerEntry>> {
+    let account = get_account(ctx, system_accounts::sequencer_stake_config_account_id())
+        .await
+        .context("Failed to read the sequencer_stake config account")?;
+    let config = sequencer_stake_core::SequencerStakeConfig::from_bytes(account.data.as_ref())
+        .context("config account data did not decode as a SequencerStakeConfig")?;
+    Ok(config.entries.get(&sequencer_key).copied())
+}

lez/sequencer/actors/executor/src/actor.rs

@@ -152,6 +152,7 @@ impl<BP: BlockPublisherTrait + Send + Sync + 'static, S: StorageActorTrait> Mess
     ) -> Self::Reply {
         // Only produce on our turn.
         if !self.sequencer.is_our_turn() {
+            self.sequencer.update_committee_absence().await;
             info!("Not our turn to produce a block, skipping");
             return Ok(());
         }
@@ -161,6 +162,7 @@ impl<BP: BlockPublisherTrait + Send + Sync + 'static, S: StorageActorTrait> Mess
         // The head rewinds under us when the sdk orphans our own unfinalized
         // blocks, and recovers once they finalize, so this is a wait.
         if let Some(high_water) = self.sequencer.rewound_below_published().await {
+            self.sequencer.update_committee_absence().await;
             warn!(
                 "Skipping turn: head rewound to {} but block {high_water} is already inscribed; \
                  waiting for the channel to restore it",

lez/sequencer/core/src/block_publisher.rs

@@ -155,8 +155,9 @@ pub trait LocalBlockPublisherTrait: Sized + Sync {
     ) -> Result<PublishOutcome>;
 
     /// Live (adopted, possibly not yet finalized) accredited-key snapshot for
-    /// this channel, read directly from the connected Bedrock node.
-    async fn accredited_keys(&self) -> Result<Vec<Ed25519PublicKey>>;
+    /// this channel with the slot its tip sits at, from one read of the
+    /// connected Bedrock node. `None` if the channel does not exist.
+    async fn accredited_keys(&self) -> Result<Option<(Vec<Ed25519PublicKey>, Slot)>>;
 
     /// Submit a committee `ChannelConfigOp` as its own, independent Mantle
     /// tx (not bundled with any block publish). `new_keys` is the full
@@ -551,14 +552,13 @@ impl BlockPublisherTrait for ZoneSdkPublisher {
             .await
     }
 
-    async fn accredited_keys(&self) -> Result<Vec<Ed25519PublicKey>> {
+    async fn accredited_keys(&self) -> Result<Option<(Vec<Ed25519PublicKey>, Slot)>> {
         Ok(self
             .node
             .channel_state(self.channel_id)
             .await
             .context("Failed to read channel state")?
-            .map(|state| state.accredited_keys.to_vec())
-            .unwrap_or_default())
+            .map(|state| (state.accredited_keys.to_vec(), state.tip_slot)))
     }
 
     async fn submit_channel_config(&self, new_keys: Vec<Ed25519PublicKey>) -> Result<()> {

lez/sequencer/core/src/committee_discovery.rs

@@ -1,8 +1,47 @@
 //! Discovery process for the `sequencer_stake` committee.
 
+use std::collections::{BTreeMap, BTreeSet};
+
 use log::warn;
 use sequencer_stake_core::{PendingUnstake, SequencerKey, SequencerStakeConfig, StakeRecord};
 
+/// The channel slot each leaving key was first seen gone from the live committee at.
+///
+/// In memory only: a restart re-stamps at the current tip and costs a pending
+/// exit one more finality delay.
+#[derive(Debug, Default)]
+pub struct CommitteeAbsence {
+    absent_since: BTreeMap<SequencerKey, u64>,
+}
+
+impl CommitteeAbsence {
+    /// Records one look at the live committee, taken with the channel tip at
+    /// `tip_slot`.
+    pub fn observe(
+        &mut self,
+        live: &[SequencerKey],
+        leaving: &BTreeSet<SequencerKey>,
+        tip_slot: u64,
+    ) {
+        self.absent_since.retain(|key, _| leaving.contains(key));
+        for key in leaving {
+            if live.contains(key) {
+                self.absent_since.remove(key);
+            } else {
+                self.absent_since.entry(*key).or_insert(tip_slot);
+            }
+        }
+    }
+
+    /// Whether the channel slot that first showed `key` gone is irreversible.
+    #[must_use]
+    pub fn removal_is_final(&self, key: SequencerKey, finalized_slot: Option<u64>) -> bool {
+        self.absent_since
+            .get(&key)
+            .is_some_and(|absent_since| finalized_slot > Some(*absent_since))
+    }
+}
+
 /// The accredited-keys list LEZ state says the channel should have, or `None`
 /// if it already matches the live Bedrock committee.
 ///
@@ -70,40 +109,45 @@ pub fn finalize_unstake_candidates(state: &lee::V03State) -> Vec<(lee::AccountId
         .collect()
 }
 
-/// Block-validity rule for a `FinalizeUnstake` on `ownership_id`.
-///
-/// A partial release is always valid: `UnstakeRequest` already guarantees it
-/// leaves the key at or above the minimum, so committee membership is
-/// unaffected. A full drain — measured against tracked stake, not balance,
-/// which anyone can inflate — is valid only once the key is no longer
-/// accredited. An unknown account, no pending request, or no config entry
-/// also counts as valid; the program itself rejects those cases anyway.
-///
-/// TODO: checks live Bedrock membership, not an ordered history walk like the
-/// spec calls for. Fine for the sequencer building the next block, but a
-/// follower re-checking an already-adopted block has no independent way to
-/// verify it this way. Switch once zone-sdk exposes ordered `ChannelConfigOp`
-/// data to LEZ.
+// TODO: Only checked on blocks we build, never re-checked on adoption.
+/// Whether a `FinalizeUnstake` on `ownership_id` may go in a block: one that
+/// removes the key waits until the removal can no longer be undone.
 #[must_use]
 pub fn finalize_unstake_is_valid(
     state: &lee::V03State,
     ownership_id: lee::AccountId,
-    live_accredited_keys: &[SequencerKey],
+    absence: &CommitteeAbsence,
+    finalized_slot: Option<u64>,
 ) -> bool {
     let Some(record) = stake_record(state, ownership_id) else {
         return true;
     };
-    let Some(pending) = record.pending_unstake else {
+    if record.pending_unstake.is_none() {
         return true;
-    };
+    }
     let Some(entry) =
         read_config(state).and_then(|config| config.entries.get(&record.sequencer_key).copied())
     else {
         return true;
     };
 
-    let fully_drains = entry.total_staked == pending.amount;
-    !fully_drains || !live_accredited_keys.contains(&record.sequencer_key)
+    let fully_drains = entry.net_stake() == 0;
+    !fully_drains || absence.removal_is_final(record.sequencer_key, finalized_slot)
+}
+
+/// Keys whose pending release takes them out of the committee.
+#[must_use]
+pub fn keys_leaving_the_committee(state: &lee::V03State) -> BTreeSet<SequencerKey> {
+    let Some(config) = read_config(state) else {
+        return BTreeSet::new();
+    };
+
+    config
+        .entries
+        .into_iter()
+        .filter(|(_, entry)| entry.net_stake() == 0)
+        .map(|(key, _)| key)
+        .collect()
 }
 
 /// Reads the `sequencer_stake` config account — a single account read, not a
@@ -349,37 +393,85 @@ mod tests {
         let staked = Staked::new(5, MINIMUM + 10).pending(10);
         let state = state_with([staked]);
 
+        // Neither still accredited nor anything finalized matters here.
         assert!(finalize_unstake_is_valid(
             &state,
             staked.account_id,
-            &[staked.key]
+            &CommitteeAbsence::default(),
+            None
         ));
-        assert!(finalize_unstake_is_valid(&state, staked.account_id, &[]));
     }
 
     #[test]
-    fn a_full_drain_is_valid_only_once_absent_from_the_live_committee() {
+    fn a_full_drain_waits_for_the_removal_to_finalize() {
         let staked = Staked::new(6, MINIMUM).pending(MINIMUM);
         let state = state_with([staked]);
+        let watched = BTreeSet::from([staked.key]);
+        let valid = |absence: &CommitteeAbsence, finalized| {
+            finalize_unstake_is_valid(&state, staked.account_id, absence, finalized)
+        };
 
-        assert!(!finalize_unstake_is_valid(
-            &state,
-            staked.account_id,
-            &[staked.key]
-        ));
-        assert!(finalize_unstake_is_valid(&state, staked.account_id, &[]));
+        // Never looked at, or looked at and still accredited.
+        let mut absence = CommitteeAbsence::default();
+        assert!(!valid(&absence, Some(100)));
+        absence.observe(&[staked.key], &watched, 10);
+        assert!(!valid(&absence, Some(100)));
+
+        // Seen leaving at height 10: only a later block finalizing frees it.
+        absence.observe(&[], &watched, 10);
+        for finalized in [None, Some(9), Some(10)] {
+            assert!(
+                !valid(&absence, finalized),
+                "not final yet at {finalized:?}"
+            );
+        }
+        assert!(valid(&absence, Some(11)));
+    }
+
+    #[test]
+    fn each_removal_only_frees_its_own_release() {
+        let key = test_key(6);
+        let watched = BTreeSet::from([key]);
+        let mut absence = CommitteeAbsence::default();
+
+        absence.observe(&[], &watched, 10);
+        assert!(absence.removal_is_final(key, Some(11)));
+
+        // Seen back in the committee: that removal no longer counts.
+        absence.observe(&[key], &watched, 20);
+        assert!(!absence.removal_is_final(key, Some(100)));
+        absence.observe(&[], &watched, 30);
+        assert!(!absence.removal_is_final(key, Some(30)));
+        assert!(absence.removal_is_final(key, Some(31)));
+
+        // FinalizeUnstake ran: nothing pending, so the note is dropped.
+        absence.observe(&[], &BTreeSet::new(), 40);
+        assert!(!absence.removal_is_final(key, Some(100)));
     }
 
     #[test]
     fn a_donated_balance_does_not_make_a_full_drain_look_partial() {
-        // The drain is measured against the tracked stake, so a donation
-        // sitting on the ownership account does not reclassify it as partial.
+        // Measured against tracked stake, so a donation cannot hide the drain.
         let staked = Staked::new(7, MINIMUM).pending(MINIMUM).donated(1);
 
         assert!(!finalize_unstake_is_valid(
             &state_with([staked]),
             staked.account_id,
-            &[staked.key]
+            &CommitteeAbsence::default(),
+            Some(100)
         ));
     }
+
+    #[test]
+    fn only_keys_a_release_would_remove_are_watched() {
+        // A key that stays in the committee has no removal to wait for.
+        let exiting = Staked::new(4, MINIMUM).pending(MINIMUM);
+        let staying = Staked::new(6, MINIMUM);
+        let shrinking = Staked::new(8, 2 * MINIMUM).pending(MINIMUM);
+
+        assert_eq!(
+            keys_leaving_the_committee(&state_with([exiting, staying, shrinking])),
+            BTreeSet::from([exiting.key])
+        );
+    }
 }

lez/sequencer/core/src/lib.rs

@@ -140,6 +140,8 @@ pub struct SequencerCore<
     watchers: TaskGroup,
     /// Channel tip slot as of the last committee-config submission.
     last_committee_submission_slot: Option<Slot>,
+    /// Gates `FinalizeUnstake` on the committee removal being final.
+    committee_absence: committee_discovery::CommitteeAbsence,
 }
 
 impl<BP: BlockPublisherTrait, S: StorageActorTrait> SequencerCore<BP, S> {
@@ -417,6 +419,7 @@ impl<BP: BlockPublisherTrait, S: StorageActorTrait> SequencerCore<BP, S> {
             block_publisher,
             watchers,
             last_committee_submission_slot: None,
+            committee_absence: committee_discovery::CommitteeAbsence::default(),
         };
 
         sequencer_core_metrics::record_chain_height(sequencer_core.chain_height().await);
@@ -676,11 +679,30 @@ impl<BP: BlockPublisherTrait, S: StorageActorTrait> SequencerCore<BP, S> {
         })
     }
 
+    /// Updates the absence record from a fresh read of the live committee, and
+    /// returns that committee. `None` if the read failed, which is not the same
+    /// as an empty committee.
+    ///
+    /// Worth calling off-turn: a key seen back in the committee resets its clock.
+    pub async fn update_committee_absence(
+        &mut self,
+    ) -> Option<Vec<sequencer_stake_core::SequencerKey>> {
+        let snapshot = self.live_accredited_sequencer_keys().await;
+        if let Some((live, tip_slot)) = snapshot.as_ref() {
+            let leaving = self
+                .with_state(committee_discovery::keys_leaving_the_committee)
+                .await;
+            self.committee_absence
+                .observe(live, &leaving, tip_slot.into_inner());
+        }
+        snapshot.map(|(keys, _)| keys)
+    }
+
     /// Runs everything this sequencer owes its turn: builds a block from
     /// mempool transactions, publishes it via zone-sdk, and submits any
     /// committee-config update the new state calls for.
     pub async fn run_production_turn(&mut self) -> Result<u64> {
-        let live_accredited_keys = self.live_accredited_sequencer_keys().await;
+        let live_accredited_keys = self.update_committee_absence().await;
 
         let BlockWithMeta {
             block,
@@ -725,12 +747,13 @@ impl<BP: BlockPublisherTrait, S: StorageActorTrait> SequencerCore<BP, S> {
     }
 
     /// Live committee snapshot for gating `FinalizeUnstake` inclusion and
-    /// committee updates. `None` if it could not be read.
+    /// committee updates, with the channel tip slot it was read at. `None` if
+    /// the channel is missing or unreadable.
     async fn live_accredited_sequencer_keys(
         &self,
-    ) -> Option<Vec<sequencer_stake_core::SequencerKey>> {
+    ) -> Option<(Vec<sequencer_stake_core::SequencerKey>, Slot)> {
         match self.block_publisher.accredited_keys().await {
-            Ok(keys) => Some(
+            Ok(Some((keys, tip_slot))) => Some((
                 keys.iter()
                     .filter_map(|key| {
                         sequencer_stake_core::SequencerKey::new(key.to_bytes()).or_else(|| {
@@ -742,7 +765,15 @@ impl<BP: BlockPublisherTrait, S: StorageActorTrait> SequencerCore<BP, S> {
                         })
                     })
                     .collect(),
-            ),
+                tip_slot,
+            )),
+            Ok(None) => {
+                warn!(
+                    "No channel to read a live committee from; skipping FinalizeUnstake inclusion \
+                     and committee updates this round"
+                );
+                None
+            }
             Err(err) => {
                 warn!(
                     "Failed to read live committee snapshot; skipping FinalizeUnstake inclusion \
@@ -975,6 +1006,7 @@ impl<BP: BlockPublisherTrait, S: StorageActorTrait> SequencerCore<BP, S> {
             mut working_state,
             pending_dispatches,
             finalize_unstake_txs,
+            committee_update,
         ) = {
             let chain = self.chain.lock().await;
             let tip = chain.head_tip();
@@ -1004,15 +1036,30 @@ impl<BP: BlockPublisherTrait, S: StorageActorTrait> SequencerCore<BP, S> {
                 }
             }
 
+            // Committee membership follows finalized state only.
+            let committee_update = live_accredited_keys
+                .and_then(|keys| committee_discovery::committee_update(chain.final_state(), keys));
+
             (
                 prev,
                 height,
                 chain.head_state().clone(),
                 pending,
                 build_finalize_unstake_txs(chain.head_state()),
+                committee_update,
             )
         };
 
+        // The last irreversible bedrock slot.
+        let finalized_slot = self
+            .store
+            .get_zone_checkpoint()
+            .await
+            .inspect_err(|err| warn!("Failed to read the zone checkpoint: {err:#}"))
+            .ok()
+            .flatten()
+            .map(|checkpoint| checkpoint.lib_slot.into_inner());
+
         if !settled.is_empty() {
             if let Err(err) = self
                 .store
@@ -1151,7 +1198,12 @@ impl<BP: BlockPublisherTrait, S: StorageActorTrait> SequencerCore<BP, S> {
             // met (mempool: whoever wants it finalized resubmits;
             // discovery-sourced: reconstructed fresh next block), so it
             // doesn't need requeuing here.
-            if !finalize_unstake_is_includable(&working_state, &tx, live_accredited_keys) {
+            if !finalize_unstake_is_includable(
+                &working_state,
+                &tx,
+                &self.committee_absence,
+                finalized_slot,
+            ) {
                 continue;
             }
 
@@ -1216,9 +1268,6 @@ impl<BP: BlockPublisherTrait, S: StorageActorTrait> SequencerCore<BP, S> {
 
         sequencer_core_metrics::record_block_creation_time(now.elapsed());
 
-        let committee_update = live_accredited_keys
-            .and_then(|keys| committee_discovery::committee_update(&working_state, keys));
-
         Ok(BlockWithMeta {
             block,
             withdrawals,
@@ -2119,16 +2168,18 @@ fn build_bridge_deposit_tx_from_event(event: &PendingDepositEventRecord) -> Resu
 fn finalize_unstake_is_includable(
     state: &lee::V03State,
     tx: &LeeTransaction,
-    live_accredited_keys: Option<&[sequencer_stake_core::SequencerKey]>,
+    committee_absence: &committee_discovery::CommitteeAbsence,
+    finalized_slot: Option<u64>,
 ) -> bool {
     let Some(ownership_id) = finalize_unstake_ownership_account(tx) else {
         return true;
     };
-    // Without a committee snapshot there is nothing to check a FinalizeUnstake
-    // against, so none is includable this block.
-    live_accredited_keys.is_some_and(|live_accredited_keys| {
-        committee_discovery::finalize_unstake_is_valid(state, ownership_id, live_accredited_keys)
-    })
+    committee_discovery::finalize_unstake_is_valid(
+        state,
+        ownership_id,
+        committee_absence,
+        finalized_slot,
+    )
 }
 
 /// The ownership account a `FinalizeUnstake` call targets, or `None` if `tx`

lez/sequencer/core/src/mock.rs

@@ -98,8 +98,8 @@ impl BlockPublisherTrait for MockBlockPublisher {
         self.publish_block(block, Vec::new()).await
     }
 
-    async fn accredited_keys(&self) -> Result<Vec<Ed25519PublicKey>> {
-        Ok(Vec::new())
+    async fn accredited_keys(&self) -> Result<Option<(Vec<Ed25519PublicKey>, Slot)>> {
+        Ok(self.tip_slot.map(|slot| (Vec::new(), slot)))
     }
 
     async fn submit_channel_config(&self, _new_keys: Vec<Ed25519PublicKey>) -> Result<()> {

lez/sequencer/core/src/tests.rs

@@ -1272,9 +1272,59 @@ async fn build_block_from_mempool() {
     assert_eq!(sequencer.chain_height().await, genesis_height);
 }
 
+#[tokio::test]
+async fn a_stake_only_moves_the_committee_once_it_has_finalized() {
+    // Genesis stakes the bootstrap key, so the head wants it accredited already.
+    let (mut sequencer, mempool_handle) = common_setup().await;
+
+    assert!(
+        sequencer
+            .build_block_from_mempool(Some(&[]))
+            .await
+            .unwrap()
+            .committee_update
+            .is_none(),
+        "an unfinalized stake must not move the committee"
+    );
+
+    let genesis = sequencer
+        .store
+        .block_at_id(lee_core::GENESIS_BLOCK_ID)
+        .await
+        .unwrap()
+        .unwrap();
+    apply_follow_update(
+        sequencer.block_store().storage_ref(),
+        &sequencer.chain(),
+        &mempool_handle,
+        FollowUpdate {
+            finalized: vec![(MsgId::from(genesis.header.hash.0), genesis)],
+            ..empty_follow_update()
+        },
+    )
+    .await;
+
+    let wanted = sequencer
+        .build_block_from_mempool(Some(&[]))
+        .await
+        .unwrap()
+        .committee_update
+        .expect("the stake is irreversible now, so the committee should follow it");
+    assert!(
+        sequencer
+            .build_block_from_mempool(Some(&wanted))
+            .await
+            .unwrap()
+            .committee_update
+            .is_none(),
+        "a committee that already matches must not be resubmitted"
+    );
+}
+
 #[test]
-fn without_a_committee_snapshot_finalize_unstake_is_not_includable() {
+fn the_committee_gate_holds_back_neither_ordinary_txs_nor_unknown_accounts() {
     let state = V03State::new();
+    let absence = crate::committee_discovery::CommitteeAbsence::default();
     let finalize_unstake = build_finalize_unstake_tx(
         AccountId::new([1; 32]),
         sequencer_stake_core::PendingUnstake {
@@ -1284,26 +1334,19 @@ fn without_a_committee_snapshot_finalize_unstake_is_not_includable() {
     )
     .expect("FinalizeUnstake tx should build");
 
-    // An empty committee is a real answer ("no key is accredited"), so it lets
-    // a full drain through. No answer at all must not.
+    // An ownership account no state knows about is left to the program to reject.
     assert!(finalize_unstake_is_includable(
         &state,
         &finalize_unstake,
-        Some(&[])
+        &absence,
+        None
     ));
-    assert!(!finalize_unstake_is_includable(
+    assert!(finalize_unstake_is_includable(
         &state,
-        &finalize_unstake,
+        &common::test_utils::produce_dummy_empty_transaction(),
+        &absence,
         None
     ));
-}
-
-#[test]
-fn a_missing_committee_snapshot_holds_back_nothing_else() {
-    let state = V03State::new();
-    let ordinary_tx = common::test_utils::produce_dummy_empty_transaction();
-
-    assert!(finalize_unstake_is_includable(&state, &ordinary_tx, None));
 }
 
 #[tokio::test]

test_fixtures/src/config.rs

@@ -332,6 +332,12 @@ fn founding_stake_owner_seed(index: usize) -> [u8; 32] {
     seed
 }
 
+/// Key owning sequencer `index`'s founding stake.
+pub fn founding_stake_owner_key(index: usize) -> Result<PrivateKey> {
+    PrivateKey::try_new(founding_stake_owner_seed(index))
+        .context("Failed to build the founding stake ownership key")
+}
+
 /// Genesis entries staking every sequencer in `sequencer_signing_keys`, so the
 /// creator opens the channel already accrediting all of them.
 pub fn genesis_sequencer_stakes(sequencer_signing_keys: &[[u8; 32]]) -> Result<Vec<GenesisAction>> {
@@ -342,8 +348,7 @@ pub fn genesis_sequencer_stakes(sequencer_signing_keys: &[[u8; 32]]) -> Result<V
             let public_key = Ed25519Key::from_bytes(signing_key).public_key();
             let sequencer_key = SequencerKey::new(public_key.to_bytes())
                 .context("Sequencer signing key is not a valid Ed25519 point")?;
-            let owner = PrivateKey::try_new(founding_stake_owner_seed(index))
-                .context("Failed to build the founding stake ownership key")?;
+            let owner = founding_stake_owner_key(index)?;
             Ok(GenesisAction::StakeSequencer {
                 sequencer_key,
                 ownership_public_key: PublicKey::new_from_private_key(&owner),