consensus: change verifier poll_ready handling.

This makes the component verifiers both always return `poll_ready`,
because they do not exert backpressure and cannot fail.

The checkpoint verifier now immediately rejects any blocks that arrive
after it finishes checkpointing, instead of marking the service itself
as failed.

The chain verifier is agnostic to the readiness behavior of its
components, and reports readiness when they are both ready.

zebra-consensus/src/chain.rs

@@ -50,11 +50,15 @@ where
     type Future =
         Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
 
-    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        // We don't expect the verifiers to exert backpressure on our
+        match (self.checkpoint.poll_ready(cx), self.block.poll_ready(cx)) {
-        // users, so we don't need to call the verifier's `poll_ready` here.
+            // First, fail if either service fails.
-        // (And we don't know which verifier to choose at this point, anyway.)
+            (Poll::Ready(Err(e)), _) | (_, Poll::Ready(Err(e))) => Poll::Ready(Err(e)),
-        Poll::Ready(Ok(()))
+            // Second, we're unready if either service is unready.
+            (Poll::Pending, _) | (_, Poll::Pending) => Poll::Pending,
+            // Finally, we're ready if both services are ready and OK.
+            (Poll::Ready(Ok(())), Poll::Ready(Ok(()))) => Poll::Ready(Ok(())),
+        }
     }
 
     fn call(&mut self, block: Arc<Block>) -> Self::Future {

zebra-consensus/src/checkpoint.rs

@@ -763,13 +763,15 @@ where
         Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
 
     fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        match self.previous_checkpoint_height() {
+        Poll::Ready(Ok(()))
-            FinalCheckpoint => Poll::Ready(Err("there are no checkpoints left to verify".into())),
-            _ => Poll::Ready(Ok(())),
-        }
     }
 
     fn call(&mut self, block: Arc<Block>) -> Self::Future {
+        // Immediately reject all incoming blocks that arrive after we've finished.
+        if let FinalCheckpoint = self.previous_checkpoint_height() {
+            return async { Err("checkpoint request after checkpointing finished".into()) }.boxed();
+        }
+
         // Queue the block for verification, until we receive all the blocks for
         // the current checkpoint range.
         let rx = self.queue_block(block.clone());