WIP: node: Update p2panda to new API #195
Conversation
jsparber
force-pushed
the
jsparber/new_p2panda
branch
from
23c36d4 to
3f9f4a6
Compare
reflection-node/src/network.rs
Outdated
| pub type TopicSyncManager = p2panda_sync::manager::TopicSyncManager< | ||
| TopicId, | ||
| p2panda_store::SqliteStore<LogId, ReflectionExtensions>, | ||
| TopicStore, | ||
| LogId, | ||
| ReflectionExtensions, | ||
| >; | ||
| pub type LogSyncError = p2panda_net::sync::LogSyncError<TopicSyncManager>; | ||
| pub type SyncHandle = p2panda_net::sync::SyncHandle<TopicSyncManager>; | ||
| pub type SyncHandleError = p2panda_net::sync::SyncHandleError<TopicSyncManager>; |
There was a problem hiding this comment.
Is this (the error types) the only place the TopicSyncManger type is needed?
We intended to abstract this type away behind LogSync provided by p2panda_net, so if it is really required here for error types maybe we can look again and try to take this requirement away.
There was a problem hiding this comment.
Yes, I made the type definition private to make clear that it's only used here.
There was a problem hiding this comment.
oh it's also used in SyncHandle
There was a problem hiding this comment.
Great, thanks, I think I see how we can remove this type requirement from the API. I'll give it a go and let you know when it's ready, should be doable before the release.
There was a problem hiding this comment.
This PR should remove the need to define the TopicSyncManager type.
jsparber
force-pushed
the
jsparber/new_p2panda
branch
from
3f9f4a6 to
dad8f49
Compare
jsparber
force-pushed
the
jsparber/new_p2panda
branch
from
dad8f49 to
1114f61
Compare
|
|
||
| let abort_handle = spawn(async move { | ||
| while let Ok(event) = topic_rx.recv().await { | ||
| while let Some(Ok(event)) = topic_rx.next().await { |
There was a problem hiding this comment.
What's the error that can happen on each item of the stream? Is the error something persistent, or can we just go to the next item in the stream when we see an error?
There was a problem hiding this comment.
Same thing for the GossipSubscription
There was a problem hiding this comment.
Looking at the code GossipSubscription error is only https://docs.rs/tokio-stream/latest/tokio_stream/wrappers/errors/enum.BroadcastStreamRecvError.html and it will be fine to keep reading after an error. I think the item stream should return the specific error not an enum of all errors.
There was a problem hiding this comment.
Now looked also at the SyncSubscription is the same as the GossipSubscription. So the only error that can happen is https://docs.rs/tokio-stream/latest/tokio_stream/wrappers/errors/enum.BroadcastStreamRecvError.html. I wonder if that error couldn't be handled internally.
There was a problem hiding this comment.
Now looked also at the
SyncSubscriptionis the same as theGossipSubscription. So the only error that can happen ishttps://docs.rs/tokio-stream/latest/tokio_stream/wrappers/errors/enum.BroadcastStreamRecvError.html. I wonder if that error couldn't be handled internally.
I'm actually touching on this area for my work to remove the unwanted trait leakage from p2panda-sync. I will make it so that BroadcastStreamReccError is returned from the sync subscription. I could do the same with gossip as well.
What did you mean by handling the error internally though?
There was a problem hiding this comment.
The error indicates that messages have been dropped, which I believe is important to know, even though the user doesn't have a chance to recover from this in the gossip case?
However, this is an issue with this particular mpmc channel implementation of tokio. We could switch to flume or something else which offers bounded channels who will instead apply backpressure rather than dropping items. I believe we can then also hide the https://docs.rs/flume/latest/flume/enum.RecvError.html error since it'll be equal to a Stream termination (yielding None). Looks like they implement it exactly like that: https://docs.rs/flume/latest/src/flume/async.rs.html#573
There was a problem hiding this comment.
What did you mean by handling the error internally though?
I mean not exposing the error to the consumer at all, and make sure internally that it doesn't happen or that it's handled. For the consumer it's quite unclear how to handle the error, is there a way to request missed operations because of the slow read?
No description provided.