diff --git a/linkerd/proxy/api-resolve/src/resolve.rs b/linkerd/proxy/api-resolve/src/resolve.rs
index 78076229a5..18823da0de 100644
--- a/linkerd/proxy/api-resolve/src/resolve.rs
+++ b/linkerd/proxy/api-resolve/src/resolve.rs
@@ -104,14 +104,10 @@ where
     }
 }
 
-impl resolve::Resolution for Resolution {
-    type Endpoint = Metadata;
-    type Error = grpc::Status;
+impl Stream for Resolution {
+    type Item = Result<resolve::Update<Metadata>, grpc::Status>;
 
-    fn poll(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Update<Self::Endpoint>, Self::Error>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let mut this = self.project();
         loop {
             match ready!(this.inner.as_mut().poll_next(cx)) {
@@ -126,7 +122,7 @@ impl resolve::Resolution for Resolution {
                             .collect::<Vec<_>>();
                         if !addr_metas.is_empty() {
                             debug!(endpoints = %addr_metas.len(), "Add");
-                            return Poll::Ready(Ok(Update::Add(addr_metas)));
+                            return Poll::Ready(Some(Ok(Update::Add(addr_metas))));
                         }
                     }
 
@@ -137,7 +133,7 @@ impl resolve::Resolution for Resolution {
                             .collect::<Vec<_>>();
                         if !sock_addrs.is_empty() {
                             debug!(endpoints = %sock_addrs.len(), "Remove");
-                            return Poll::Ready(Ok(Update::Remove(sock_addrs)));
+                            return Poll::Ready(Some(Ok(Update::Remove(sock_addrs))));
                         }
                     }
 
@@ -148,14 +144,12 @@ impl resolve::Resolution for Resolution {
                         } else {
                             Update::DoesNotExist
                         };
-                        return Poll::Ready(Ok(update.into()));
+                        return Poll::Ready(Some(Ok(update.into())));
                     }
 
                     None => {} // continue
                 },
-                None => {
-                    return Poll::Ready(Err(grpc::Status::new(grpc::Code::Ok, "end of stream")))
-                }
+                None => return Poll::Ready(None),
             };
         }
     }
diff --git a/linkerd/proxy/core/src/lib.rs b/linkerd/proxy/core/src/lib.rs
index 7102ead4cb..0e5a43e03b 100644
--- a/linkerd/proxy/core/src/lib.rs
+++ b/linkerd/proxy/core/src/lib.rs
@@ -2,4 +2,4 @@
 
 pub mod resolve;
 
-pub use self::resolve::{Resolution, Resolve};
+pub use self::{resolve::Resolve, resolve::Update};
diff --git a/linkerd/proxy/core/src/resolve.rs b/linkerd/proxy/core/src/resolve.rs
index 7fe22fb5ef..16f9362fba 100644
--- a/linkerd/proxy/core/src/resolve.rs
+++ b/linkerd/proxy/core/src/resolve.rs
@@ -1,14 +1,14 @@
+use futures::stream::TryStream;
 use linkerd2_error::Error;
 use std::future::Future;
 use std::net::SocketAddr;
-use std::pin::Pin;
 use std::task::{Context, Poll};
 
-/// Resolves `T`-typed names/addresses as a `Resolution`.
+/// Resolves `T`-typed names/addresses as an infinite stream of `Update<Self::Endpoint>`.
 pub trait Resolve<T> {
     type Endpoint;
     type Error: Into<Error>;
-    type Resolution: Resolution<Endpoint = Self::Endpoint>;
+    type Resolution: TryStream<Ok = Update<Self::Endpoint>, Error = Self::Error>;
     type Future: Future<Output = Result<Self::Resolution, Self::Error>>;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
@@ -23,27 +23,6 @@ pub trait Resolve<T> {
     }
 }
 
-/// An infinite stream of endpoint updates.
-pub trait Resolution {
-    type Endpoint;
-    type Error: Into<Error>;
-
-    fn poll(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Update<Self::Endpoint>, Self::Error>>;
-
-    fn poll_unpin(
-        &mut self,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Update<Self::Endpoint>, Self::Error>>
-    where
-        Self: Unpin,
-    {
-        Pin::new(self).poll(cx)
-    }
-}
-
 #[derive(Clone, Debug)]
 pub struct Service<S>(S);
 
@@ -57,13 +36,13 @@ pub enum Update<T> {
 
 // === impl Resolve ===
 
-impl<S, T, R> Resolve<T> for S
+impl<S, T, R, E> Resolve<T> for S
 where
     S: tower::Service<T, Response = R>,
     S::Error: Into<Error>,
-    R: Resolution,
+    R: TryStream<Ok = Update<E>, Error = S::Error>,
 {
-    type Endpoint = <R as Resolution>::Endpoint;
+    type Endpoint = E;
     type Error = S::Error;
     type Resolution = S::Response;
     type Future = S::Future;
diff --git a/linkerd/proxy/discover/src/from_resolve.rs b/linkerd/proxy/discover/src/from_resolve.rs
index a031b0fc88..feb8370864 100644
--- a/linkerd/proxy/discover/src/from_resolve.rs
+++ b/linkerd/proxy/discover/src/from_resolve.rs
@@ -1,6 +1,6 @@
-use futures::{ready, Stream, TryFuture};
+use futures::{ready, Stream, TryFuture, TryStream};
 use indexmap::IndexSet;
-use linkerd2_proxy_core::resolve::{Resolution, Resolve, Update};
+use linkerd2_proxy_core::resolve::{Resolve, Update};
 use pin_project::pin_project;
 use std::collections::VecDeque;
 use std::future::Future;
@@ -10,45 +10,50 @@ use std::task::{Context, Poll};
 use tower::discover::Change;
 
 #[derive(Clone, Debug)]
-pub struct FromResolve<R> {
+pub struct FromResolve<R, E> {
     resolve: R,
+    _marker: std::marker::PhantomData<fn(E)>,
 }
 
 #[pin_project]
 #[derive(Debug)]
-pub struct DiscoverFuture<F> {
+pub struct DiscoverFuture<F, E> {
     #[pin]
     future: F,
+    _marker: std::marker::PhantomData<fn(E)>,
 }
 
 /// Observes an `R`-typed resolution stream, using an `M`-typed endpoint stack to
 /// build a service for each endpoint.
 #[pin_project]
-pub struct Discover<R: Resolution> {
+pub struct Discover<R: TryStream, E> {
     #[pin]
     resolution: R,
     active: IndexSet<SocketAddr>,
-    pending: VecDeque<Change<SocketAddr, R::Endpoint>>,
+    pending: VecDeque<Change<SocketAddr, E>>,
 }
 
 // === impl FromResolve ===
 
-impl<R> FromResolve<R> {
+impl<R, E> FromResolve<R, E> {
     pub fn new<T>(resolve: R) -> Self
     where
         R: Resolve<T>,
     {
-        Self { resolve }
+        Self {
+            resolve,
+            _marker: std::marker::PhantomData,
+        }
     }
 }
 
-impl<T, R> tower::Service<T> for FromResolve<R>
+impl<T, R, E> tower::Service<T> for FromResolve<R, E>
 where
     R: Resolve<T> + Clone,
 {
-    type Response = Discover<R::Resolution>;
+    type Response = Discover<R::Resolution, E>;
     type Error = R::Error;
-    type Future = DiscoverFuture<R::Future>;
+    type Future = DiscoverFuture<R::Future, E>;
 
     #[inline]
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
@@ -59,18 +64,19 @@ where
     fn call(&mut self, target: T) -> Self::Future {
         Self::Future {
             future: self.resolve.resolve(target),
+            _marker: std::marker::PhantomData,
         }
     }
 }
 
 // === impl DiscoverFuture ===
 
-impl<F> Future for DiscoverFuture<F>
+impl<F, E> Future for DiscoverFuture<F, E>
 where
     F: TryFuture,
-    F::Ok: Resolution,
+    F::Ok: TryStream,
 {
-    type Output = Result<Discover<F::Ok>, F::Error>;
+    type Output = Result<Discover<F::Ok, E>, F::Error>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let resolution = ready!(self.project().future.try_poll(cx))?;
@@ -80,7 +86,7 @@ where
 
 // === impl Discover ===
 
-impl<R: Resolution> Discover<R> {
+impl<R: TryStream, E> Discover<R, E> {
     pub fn new(resolution: R) -> Self {
         Self {
             resolution,
@@ -90,8 +96,11 @@ impl<R: Resolution> Discover<R> {
     }
 }
 
-impl<R: Resolution> Stream for Discover<R> {
-    type Item = Result<Change<SocketAddr, R::Endpoint>, R::Error>;
+impl<R, E> Stream for Discover<R, E>
+where
+    R: TryStream<Ok = Update<E>>,
+{
+    type Item = Result<Change<SocketAddr, E>, R::Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         loop {
@@ -100,24 +109,27 @@ impl<R: Resolution> Stream for Discover<R> {
                 return Poll::Ready(Some(Ok(change)));
             }
 
-            match ready!(this.resolution.poll(cx))? {
-                Update::Add(endpoints) => {
-                    for (addr, endpoint) in endpoints.into_iter() {
-                        this.active.insert(addr);
-                        this.pending.push_back(Change::Insert(addr, endpoint));
+            match ready!(this.resolution.try_poll_next(cx)) {
+                Some(update) => match update? {
+                    Update::Add(endpoints) => {
+                        for (addr, endpoint) in endpoints.into_iter() {
+                            this.active.insert(addr);
+                            this.pending.push_back(Change::Insert(addr, endpoint));
+                        }
                     }
-                }
-                Update::Remove(addrs) => {
-                    for addr in addrs.into_iter() {
-                        if this.active.remove(&addr) {
-                            this.pending.push_back(Change::Remove(addr));
+                    Update::Remove(addrs) => {
+                        for addr in addrs.into_iter() {
+                            if this.active.remove(&addr) {
+                                this.pending.push_back(Change::Remove(addr));
+                            }
                         }
                     }
-                }
-                Update::DoesNotExist | Update::Empty => {
-                    this.pending
-                        .extend(this.active.drain(..).map(Change::Remove));
-                }
+                    Update::DoesNotExist | Update::Empty => {
+                        this.pending
+                            .extend(this.active.drain(..).map(Change::Remove));
+                    }
+                },
+                None => return Poll::Ready(None),
             }
         }
     }
diff --git a/linkerd/proxy/discover/src/lib.rs b/linkerd/proxy/discover/src/lib.rs
index 798fde7ff3..cedc7faea7 100644
--- a/linkerd/proxy/discover/src/lib.rs
+++ b/linkerd/proxy/discover/src/lib.rs
@@ -43,7 +43,7 @@ where
     T: fmt::Display,
     R: Resolve<T> + Send + Clone + 'static,
     R::Error: Into<Error>,
-    R::Endpoint: fmt::Debug + Clone + PartialEq + Send,
+    R::Endpoint: fmt::Debug + Clone + PartialEq + Send + 'static,
     R::Resolution: Send + 'static,
     R::Future: Send + 'static,
     M: tower::Service<R::Endpoint> + Clone + Send + 'static,
@@ -51,7 +51,7 @@ where
     M::Response: Send + 'static,
     M::Future: Send + 'static,
 {
-    type Service = Buffer<MakeEndpoint<FromResolve<R>, M>>;
+    type Service = Buffer<MakeEndpoint<FromResolve<R, R::Endpoint>, M>>;
 
     fn layer(&self, make_endpoint: M) -> Self::Service {
         let make_discover =
diff --git a/linkerd/proxy/discover/src/make_endpoint.rs b/linkerd/proxy/discover/src/make_endpoint.rs
index 6d44e30fb7..ae67f80410 100644
--- a/linkerd/proxy/discover/src/make_endpoint.rs
+++ b/linkerd/proxy/discover/src/make_endpoint.rs
@@ -195,18 +195,17 @@ where
             // services. Don't process any updates until we can do so.
             ready!(this.make_endpoint.poll_ready(cx)).map_err(Into::into)?;
 
-            match ready!(this.discover.poll_discover(cx))
-                .expect("XXX(eliza): can this ever be none???")
-                .map_err(Into::into)?
-            {
-                Change::Insert(key, target) => {
-                    // Start building the service and continue. If a pending
-                    // service exists for this addr, it will be canceled.
-                    let fut = this.make_endpoint.call(target);
-                    this.make_futures.push(key, fut);
-                }
-                Change::Remove(key) => {
-                    this.pending_removals.push(key);
+            if let Some(change) = ready!(this.discover.poll_discover(cx)) {
+                match change.map_err(Into::into)? {
+                    Change::Insert(key, target) => {
+                        // Start building the service and continue. If a pending
+                        // service exists for this addr, it will be canceled.
+                        let fut = this.make_endpoint.call(target);
+                        this.make_futures.push(key, fut);
+                    }
+                    Change::Remove(key) => {
+                        this.pending_removals.push(key);
+                    }
                 }
             }
         }
diff --git a/linkerd/proxy/resolve/src/make_unpin.rs b/linkerd/proxy/resolve/src/make_unpin.rs
index fef01e3aa5..fad1a82e6c 100644
--- a/linkerd/proxy/resolve/src/make_unpin.rs
+++ b/linkerd/proxy/resolve/src/make_unpin.rs
@@ -1,5 +1,6 @@
+use futures::stream::{Stream, TryStream};
 use futures::TryFuture;
-use linkerd2_proxy_core::resolve::{self, Resolution, Update};
+use linkerd2_proxy_core::resolve;
 use pin_project::pin_project;
 use std::future::Future;
 use std::pin::Pin;
@@ -38,15 +39,11 @@ impl<T> MakeUnpin<T> {
     }
 }
 
-impl<T: Resolution> Resolution for MakeUnpin<T> {
-    type Endpoint = T::Endpoint;
-    type Error = T::Error;
+impl<T: TryStream> Stream for MakeUnpin<T> {
+    type Item = Result<T::Ok, T::Error>;
 
-    fn poll(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Update<Self::Endpoint>, Self::Error>> {
-        self.project().0.as_mut().as_mut().poll(cx)
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.project().0.as_mut().as_mut().try_poll_next(cx)
     }
 }
 
diff --git a/linkerd/proxy/resolve/src/map_endpoint.rs b/linkerd/proxy/resolve/src/map_endpoint.rs
index 6032ec93f2..5b31f74bf1 100644
--- a/linkerd/proxy/resolve/src/map_endpoint.rs
+++ b/linkerd/proxy/resolve/src/map_endpoint.rs
@@ -1,6 +1,9 @@
 //! A middleware that wraps `Resolutions`, modifying their endpoint type.
 
+use futures::stream::Stream;
+use futures::stream::TryStream;
 use futures::{ready, TryFuture};
+use linkerd2_error::Error;
 use linkerd2_proxy_core::resolve;
 use pin_project::pin_project;
 use std::future::Future;
@@ -30,11 +33,12 @@ pub struct ResolveFuture<T, F, M> {
 
 #[pin_project]
 #[derive(Clone, Debug)]
-pub struct Resolution<T, M, R> {
+pub struct Resolution<T, M, R, E> {
     #[pin]
     resolution: R,
     target: T,
     map: M,
+    _marker: std::marker::PhantomData<fn(E)>,
 }
 
 // === impl Resolve ===
@@ -54,7 +58,7 @@ where
     R: resolve::Resolve<T>,
     M: MapEndpoint<T, R::Endpoint> + Clone,
 {
-    type Response = Resolution<T, M, R::Resolution>;
+    type Response = Resolution<T, M, R::Resolution, R::Endpoint>;
     type Error = R::Error;
     type Future = ResolveFuture<T, R::Future, M>;
 
@@ -76,13 +80,14 @@ where
 
 // === impl ResolveFuture ===
 
-impl<T, F, M> Future for ResolveFuture<T, F, M>
+impl<T, F, M, E> Future for ResolveFuture<T, F, M>
 where
     F: TryFuture,
-    F::Ok: resolve::Resolution,
-    M: MapEndpoint<T, <F::Ok as resolve::Resolution>::Endpoint>,
+    F::Ok: TryStream<Ok = resolve::Update<E>>,
+    <F::Ok as TryStream>::Error: Into<Error>,
+    M: MapEndpoint<T, E>,
 {
-    type Output = Result<Resolution<T, M, F::Ok>, F::Error>;
+    type Output = Result<Resolution<T, M, F::Ok, E>, F::Error>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = self.project();
@@ -93,40 +98,41 @@ where
             resolution,
             target,
             map,
+            _marker: std::marker::PhantomData,
         }))
     }
 }
 
 // === impl Resolution ===
 
-impl<T, M, R> resolve::Resolution for Resolution<T, M, R>
+impl<T, M, R, E> Stream for Resolution<T, M, R, E>
 where
-    R: resolve::Resolution,
-    M: MapEndpoint<T, R::Endpoint>,
+    R: TryStream<Ok = resolve::Update<E>>,
+    R::Error: Into<Error>,
+    M: MapEndpoint<T, E>,
 {
-    type Endpoint = M::Out;
-    type Error = R::Error;
+    type Item = Result<resolve::Update<M::Out>, R::Error>;
 
-    fn poll(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<resolve::Update<M::Out>, Self::Error>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let this = self.project();
-        let update = match ready!(this.resolution.poll(cx))? {
-            resolve::Update::Add(eps) => {
-                let mut update = Vec::new();
-                for (a, ep) in eps.into_iter() {
-                    let ep = this.map.map_endpoint(&this.target, a, ep);
-                    update.push((a, ep));
+        let update = match ready!(this.resolution.try_poll_next(cx)) {
+            Some(result) => match result? {
+                resolve::Update::Add(eps) => {
+                    let mut update = Vec::new();
+                    for (a, ep) in eps.into_iter() {
+                        let ep = this.map.map_endpoint(&this.target, a, ep);
+                        update.push((a, ep));
+                    }
+
+                    resolve::Update::Add(update)
                 }
-
-                resolve::Update::Add(update)
-            }
-            resolve::Update::Remove(addrs) => resolve::Update::Remove(addrs),
-            resolve::Update::DoesNotExist => resolve::Update::DoesNotExist,
-            resolve::Update::Empty => resolve::Update::Empty,
+                resolve::Update::Remove(addrs) => resolve::Update::Remove(addrs),
+                resolve::Update::DoesNotExist => resolve::Update::DoesNotExist,
+                resolve::Update::Empty => resolve::Update::Empty,
+            },
+            None => return Poll::Ready(None),
         };
-        Poll::Ready(Ok(update))
+        Poll::Ready(Some(Ok(update)))
     }
 }
 
diff --git a/linkerd/proxy/resolve/src/recover.rs b/linkerd/proxy/resolve/src/recover.rs
index 0c562bfd20..0db7e2b66b 100644
--- a/linkerd/proxy/resolve/src/recover.rs
+++ b/linkerd/proxy/resolve/src/recover.rs
@@ -1,15 +1,19 @@
 //! A middleware that recovers a resolution after some failures.
 
-use futures::{prelude::*, ready};
+use futures::stream::TryStream;
+use futures::{prelude::*, ready, FutureExt, Stream};
 use indexmap::IndexMap;
 use linkerd2_error::{Error, Recover};
-use linkerd2_proxy_core::resolve::{self, Resolution as _, Update};
+use linkerd2_proxy_core::resolve::{self, Update};
 use pin_project::pin_project;
 use std::future::Future;
 use std::net::SocketAddr;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
+#[derive(Clone, Debug)]
+pub struct Eos(());
+
 #[derive(Clone, Debug)]
 pub struct Resolve<E, R> {
     resolve: R,
@@ -42,7 +46,7 @@ struct Cache<T> {
 }
 
 #[pin_project]
-enum State<F, R: resolve::Resolution, B> {
+enum State<F, R: TryStream, B> {
     Disconnected {
         backoff: Option<B>,
     },
@@ -62,7 +66,7 @@ enum State<F, R: resolve::Resolution, B> {
     Connected {
         #[pin]
         resolution: R,
-        initial: Option<Update<R::Endpoint>>,
+        initial: Option<R::Ok>,
     },
 
     Recover {
@@ -151,7 +155,7 @@ where
 
 // === impl Resolution ===
 
-impl<T, E, R> resolve::Resolution for Resolution<T, E, R>
+impl<T, E, R> Stream for Resolution<T, E, R>
 where
     T: Clone,
     R: resolve::Resolve<T>,
@@ -161,20 +165,16 @@ where
     E: Recover,
     E::Backoff: Unpin,
 {
-    type Endpoint = R::Endpoint;
-    type Error = Error;
+    type Item = Result<Update<R::Endpoint>, Error>;
 
-    fn poll(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Update<Self::Endpoint>, Self::Error>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let mut this = self.project();
         loop {
             // If a reconciliation update is buffered (i.e. after
             // reconcile_after_reconnect), process it immediately.
             if let Some(update) = this.reconcile.take() {
                 this.update_active(&update);
-                return Poll::Ready(Ok(update));
+                return Poll::Ready(Some(Ok(update)));
             }
 
             match this.inner.state {
@@ -195,24 +195,26 @@ where
                         {
                             *this.reconcile = reconcile;
                             this.update_active(&update);
-                            return Poll::Ready(Ok(update));
+                            return Poll::Ready(Some(Ok(update)));
                         }
                     }
 
                     // Process the resolution stream, updating the cache.
                     //
                     // Attempt recovery/backoff if the resolution fails.
-                    match ready!(resolution.poll_unpin(cx)) {
-                        Ok(update) => {
+
+                    match ready!(resolution.try_poll_next_unpin(cx)) {
+                        Some(Ok(update)) => {
                             this.update_active(&update);
-                            return Poll::Ready(Ok(update));
+                            return Poll::Ready(Some(Ok(update)));
                         }
-                        Err(e) => {
+                        Some(Err(e)) => {
                             this.inner.state = State::Recover {
                                 error: Some(e.into()),
                                 backoff: None,
                             }
                         }
+                        None => return Poll::Ready(None),
                     }
                 }
                 // XXX(eliza): note that this match was originally an `if let`,
@@ -300,18 +302,26 @@ where
                 State::Pending {
                     ref mut resolution,
                     ref mut backoff,
-                } => match ready!(resolution.as_mut().expect("illegal state").poll_unpin(cx)) {
-                    Err(e) => State::Recover {
+                } => match ready!(resolution
+                    .as_mut()
+                    .expect("illegal state")
+                    .try_poll_next_unpin(cx))
+                {
+                    Some(Err(e)) => State::Recover {
                         error: Some(e.into()),
                         backoff: backoff.take(),
                     },
-                    Ok(initial) => {
+                    Some(Ok(initial)) => {
                         tracing::trace!("connected");
                         State::Connected {
                             resolution: resolution.take().expect("illegal state"),
                             initial: Some(initial),
                         }
                     }
+                    None => State::Recover {
+                        error: Some(Eos(()).into()),
+                        backoff: backoff.take(),
+                    },
                 },
 
                 State::Connected { .. } => return Poll::Ready(Ok(())),
@@ -394,6 +404,14 @@ fn reconcile_after_connect<E: PartialEq>(
     }
 }
 
+impl std::fmt::Display for Eos {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "end of stream reached")
+    }
+}
+
+impl std::error::Error for Eos {}
+
 #[cfg(test)]
 mod tests {
     use super::*;