[consensus/simplex] infer ancestor certifications from notarization

When a notarization for view V arrives, the f+1 signers must have
certified V's parent before voting. Walk the ancestor chain and infer
certifications for any ancestors that are in Ready state, notifying
the resolver and reporter for each.

Fixes sync ordering (single sync_all rather than one fsync per
ancestor), avoids partial state mutation in certified, and handles
replay safely: inference is skipped during journal replay and deferred
to a post-replay pass once all Artifact::Certification entries are
applied, preventing a panic when an ancestor was certified false by
the automaton.

Author: 0xAysh

consensus/src/simplex/actors/voter/actor.rs

@@ -251,6 +251,29 @@ impl<
         }
     }
 
+    /// Syncs all open journal sections.
+    async fn sync_all_journal(&mut self) {
+        if let Some(journal) = self.journal.as_mut() {
+            journal.sync_all().await.expect("unable to sync journal");
+        }
+    }
+
+    /// Records a notarization during journal replay without running ancestor inference.
+    ///
+    /// Inference is deferred to a post-replay pass (see `run`) so that
+    /// Artifact::Certification entries already in the journal are applied before any
+    /// inference fires. Running inference here would certify ancestors as true before
+    /// their journal entry (which may say false) is replayed, causing a conflict panic.
+    async fn record_notarization(&mut self, notarization: Notarization<S, D>) {
+        let view = notarization.view();
+        let artifact = Artifact::Notarization(notarization.clone());
+        let (added, equivocator) = self.state.add_notarization(notarization);
+        if added {
+            self.append_journal(view, artifact).await;
+        }
+        self.block_equivocator(equivocator).await;
+    }
+
     /// Send a vote to every peer.
     async fn broadcast_vote<T: Sender>(
         &mut self,
@@ -414,33 +437,57 @@ impl<
     }
 
     /// Records a notarization certificate and blocks any equivocating leader.
-    async fn handle_notarization(&mut self, notarization: Notarization<S, D>) {
+    /// Returns the notarizations of any ancestors inferred as certified so callers
+    /// can forward them to the resolver and reporter.
+    async fn handle_notarization(
+        &mut self,
+        notarization: Notarization<S, D>,
+    ) -> Vec<Notarization<S, D>> {
         let view = notarization.view();
         let artifact = Artifact::Notarization(notarization.clone());
         let (added, equivocator) = self.state.add_notarization(notarization);
+        let mut inferred = Vec::new();
         if added {
             self.append_journal(view, artifact).await;
+            // The f+1 signers of `view` must have certified the parent before voting,
+            // and by the same logic the grandparent must have been certified before
+            // the parent was proposed. Walk the ancestor chain until we reach a view
+            // that is already certified, finalized, or has no notarization.
+            let mut current = view;
+            while let Some(ancestor) = self.state.infer_parent(current) {
+                if let Some(n) = self.handle_certification(ancestor, true).await {
+                    inferred.push(n);
+                }
+                current = ancestor;
+            }
+            // Sync the notarization section first so it is durable before any ancestor
+            // certification sections. This ordering ensures crash recovery always sees the
+            // notarization that triggered the inferred certifications.
+            self.sync_journal(view).await;
+            if !inferred.is_empty() {
+                // Sync all ancestor certification sections concurrently in a single call
+                // rather than one fsync per ancestor.
+                self.sync_all_journal().await;
+            }
         }
         self.block_equivocator(equivocator).await;
+        inferred
     }
 
     /// Handles the certification of a proposal.
     ///
     /// The certification may succeed, in which case the proposal can be used in future views—
     /// or fail, in which case we should nullify the view as fast as possible.
+    ///
+    /// Appends to the journal but does not sync; callers are responsible for syncing.
     async fn handle_certification(
         &mut self,
         view: View,
         success: bool,
     ) -> Option<Notarization<S, D>> {
-        // Get the notarization before advancing state
         let notarization = self.state.certified(view, success)?;
-
-        // Persist certification result for recovery
         let artifact = Artifact::Certification(Rnd::new(self.state.epoch(), view), success);
         self.append_journal(view, artifact.clone()).await;
-        self.sync_journal(view).await;
-
         Some(notarization)
     }
 
@@ -515,9 +562,13 @@ impl<
                 .await;
         }
         // Update our local round with the certificate.
-        self.handle_notarization(notarization.clone()).await;
-        // Persist the certificate before informing others.
-        self.sync_journal(view).await;
+        // handle_notarization syncs all written sections before returning.
+        let inferred = self.handle_notarization(notarization.clone()).await;
+        for n in inferred {
+            resolver.updated(Certificate::Notarization(n.clone())).await;
+            resolver.certified(n.view(), true).await;
+            self.reporter.report(Activity::Certification(n)).await;
+        }
         // Broadcast the notarization certificate
         debug!(proposal=?notarization.proposal, "broadcasting notarization");
         self.broadcast_certificate(
@@ -744,7 +795,7 @@ impl<
                         self.reporter.report(Activity::Notarize(notarize)).await;
                     }
                     Artifact::Notarization(notarization) => {
-                        self.handle_notarization(notarization.clone()).await;
+                        self.record_notarization(notarization.clone()).await;
                         resolver
                             .updated(Certificate::Notarization(notarization.clone()))
                             .await;
@@ -796,6 +847,29 @@ impl<
         }
         self.journal = Some(journal);
 
+        // Run one inference pass now that all journal entries are applied and their
+        // certification states are authoritative. This recovers certifications that
+        // were inferred during the previous run but whose Artifact::Certification
+        // journal entries were not synced before a crash.
+        let infer_views = self.state.views_with_notarization();
+        let mut any_inferred = false;
+        for view in infer_views.into_iter().rev() {
+            let mut current = view;
+            while let Some(ancestor) = self.state.infer_parent(current) {
+                let Some(n) = self.handle_certification(ancestor, true).await else {
+                    break;
+                };
+                resolver.updated(Certificate::Notarization(n.clone())).await;
+                resolver.certified(n.view(), true).await;
+                self.reporter.report(Activity::Certification(n)).await;
+                any_inferred = true;
+                current = ancestor;
+            }
+        }
+        if any_inferred {
+            self.sync_all_journal().await;
+        }
+
         // Log current view after recovery
         let end = self.context.current();
         let elapsed = end.duration_since(start).unwrap_or_default();
@@ -957,6 +1031,7 @@ impl<
                         else {
                             continue;
                         };
+                        self.sync_journal(view).await;
                         // Always forward certification outcomes to resolver.
                         // This can happen after a nullification for the same view because
                         // certification is asynchronous; finalization is the boundary that
@@ -1005,7 +1080,12 @@ impl<
                         match certificate {
                             Certificate::Notarization(notarization) => {
                                 trace!(%view, from_resolver, "received notarization");
-                                self.handle_notarization(notarization).await;
+                                let inferred = self.handle_notarization(notarization).await;
+                                for n in inferred {
+                                    resolver.updated(Certificate::Notarization(n.clone())).await;
+                                    resolver.certified(n.view(), true).await;
+                                    self.reporter.report(Activity::Certification(n)).await;
+                                }
                                 if from_resolver {
                                     resolved = Resolved::Notarization;
                                 }

consensus/src/simplex/actors/voter/mod.rs

@@ -6834,4 +6834,105 @@ mod tests {
         first_view_progress_without_timeout::<_, _, RoundRobin>(ed25519::fixture);
         first_view_progress_without_timeout::<_, _, RoundRobin>(secp256r1::fixture);
     }
+
+    /// Verifies that certifying a view sends `resolver.certified(view, true)` and
+    /// `Activity::Certification` to the reporter. This covers the notification
+    /// infrastructure shared with the infer-path; the state-level infer logic is
+    /// exercised by the `infer_parent_*` unit tests in state.rs.
+    fn certification_notifies_resolver_and_reporter<S, F, L>(mut fixture: F)
+    where
+        S: Scheme<Sha256Digest, PublicKey = PublicKey>,
+        F: FnMut(&mut deterministic::Context, &[u8], u32) -> Fixture<S>,
+        L: ElectorConfig<S>,
+    {
+        let n = 5;
+        let q = quorum(n);
+        let namespace = b"cert_notifies_resolver_reporter".to_vec();
+        let executor = deterministic::Runner::timed(Duration::from_secs(10));
+        executor.start(|mut context| async move {
+            let Fixture {
+                participants,
+                schemes,
+                ..
+            } = fixture(&mut context, &namespace, n);
+
+            let oracle =
+                start_test_network_with_peers(context.clone(), participants.clone(), false).await;
+
+            let elector = L::default();
+            let (mut mailbox, mut batcher_receiver, mut resolver_receiver, _, reporter) =
+                setup_voter(
+                    &mut context,
+                    &oracle,
+                    &participants,
+                    &schemes,
+                    elector,
+                    Duration::from_secs(30),
+                    Duration::from_secs(30),
+                    Duration::from_secs(30),
+                    mocks::application::Certifier::Always,
+                )
+                .await;
+
+            // Respond to initial batcher Update so the voter can proceed.
+            match batcher_receiver.recv().await.unwrap() {
+                batcher::Message::Update {
+                    current, response, ..
+                } => {
+                    assert_eq!(current, View::new(1));
+                    response.send(None).unwrap();
+                }
+                _ => panic!("unexpected batcher message"),
+            }
+
+            // Build a notarization for view 1 (parent = genesis).
+            let target_view = View::new(1);
+            let payload = Sha256::hash(b"cert_notifies_v1");
+            let proposal = Proposal::new(
+                Round::new(Epoch::new(333), target_view),
+                View::zero(),
+                payload,
+            );
+            let (_, notarization) = build_notarization(&schemes, &proposal, q);
+            mailbox
+                .recovered(Certificate::Notarization(notarization.clone()))
+                .await;
+
+            // Wait for Certified { view: 1, success: true } from the resolver.
+            // The voter also sends Certificate(Notarization) from try_broadcast_notarization
+            // before the automaton responds; those are ignored here.
+            loop {
+                match resolver_receiver.recv().await.unwrap() {
+                    MailboxMessage::Certified { view, success } if view == target_view => {
+                        assert!(success, "certification must succeed");
+                        break;
+                    }
+                    _ => continue,
+                }
+            }
+
+            // reporter.report runs synchronously before on_end, so reporter is
+            // guaranteed to be updated by the time Certified arrives in the channel.
+            assert!(reporter.certified.lock().contains(&target_view));
+            assert!(reporter.notarizations.lock().contains_key(&target_view));
+        });
+    }
+
+    #[test_traced]
+    fn test_certification_notifies_resolver_and_reporter() {
+        certification_notifies_resolver_and_reporter::<_, _, Random>(
+            bls12381_threshold_vrf::fixture::<MinPk, _>,
+        );
+        certification_notifies_resolver_and_reporter::<_, _, Random>(
+            bls12381_threshold_vrf::fixture::<MinSig, _>,
+        );
+        certification_notifies_resolver_and_reporter::<_, _, RoundRobin>(
+            bls12381_multisig::fixture::<MinPk, _>,
+        );
+        certification_notifies_resolver_and_reporter::<_, _, RoundRobin>(
+            bls12381_multisig::fixture::<MinSig, _>,
+        );
+        certification_notifies_resolver_and_reporter::<_, _, RoundRobin>(ed25519::fixture);
+        certification_notifies_resolver_and_reporter::<_, _, RoundRobin>(secp256r1::fixture);
+    }
 }

consensus/src/simplex/actors/voter/round.rs

@@ -219,6 +219,12 @@ impl<S: Scheme, D: Digest> Round<S, D> {
         matches!(self.certify, CertifyState::Certified(true))
     }
 
+    /// Returns true if the round has a notarization and no certification has been
+    /// initiated or completed.
+    pub const fn is_certify_ready(&self) -> bool {
+        matches!(self.certify, CertifyState::Ready) && self.notarization.is_some()
+    }
+
     /// Returns true if certification was aborted due to finalization.
     #[cfg(test)]
     pub const fn is_certify_aborted(&self) -> bool {