Implement snapshot on term end

Author: foriequal0

core/src/consensus/tendermint/engine.rs

@@ -325,7 +325,7 @@ impl ConsensusEngine for Tendermint {
     }
 }
 
-fn is_term_changed(header: &Header, parent: &Header, term_seconds: u64) -> bool {
+pub(crate) fn is_term_changed(header: &Header, parent: &Header, term_seconds: u64) -> bool {
     if term_seconds == 0 {
         return false
     }

core/src/consensus/tendermint/worker.rs

@@ -1637,6 +1637,28 @@ impl Worker {
             }
         }
 
+        let mut last_term_end = None;
+        for block_hash in &enacted {
+            let header = c.block_header(&BlockId::Hash(*block_hash)).expect("Block is enacted").decode();
+            if header.number() == 0 {
+                continue
+            }
+            let parent_header =
+                c.block_header(&BlockId::Hash(*header.parent_hash())).expect("Parent block should be enacted").decode();
+            let term_seconds = if let Some(p) = c.term_common_params(parent_header.hash().into()) {
+                p.term_seconds()
+            } else {
+                continue
+            };
+            if super::engine::is_term_changed(&header, &parent_header, term_seconds) {
+                last_term_end = Some(*block_hash);
+            }
+        }
+        if let Some(last_term_end) = last_term_end {
+            // TODO: Reduce the snapshot frequency.
+            self.snapshot_notify_sender.notify(last_term_end);
+        }
+
         if let Some((last, rest)) = imported.split_last() {
             let (imported, last_proposal_header) = {
                 let header =

sync/src/snapshot/mod.rs

@@ -57,6 +57,7 @@ impl Service {
                 } else {
                     cinfo!(SYNC, "Snapshot is ready for block: {}", block_hash)
                 }
+                // TODO: Prune old snapshots
             }
             cinfo!(SYNC, "Snapshot service is stopped")
         });