feat: Add dynamic sync mode to peer_network_interface #394
Conversation
Signed-off-by: William Hankins <[email protected]>
Signed-off-by: William Hankins <[email protected]>
Signed-off-by: William Hankins <[email protected]>
Signed-off-by: William Hankins <[email protected]>
Signed-off-by: William Hankins <[email protected]>
Signed-off-by: William Hankins <[email protected]>
Signed-off-by: William Hankins <[email protected]>
common/src/commands/sync.rs
Outdated
| pub enum SyncCommand { | ||
| ChangeSyncPoint { slot: Slot, hash: BlockHash }, | ||
| } |
There was a problem hiding this comment.
| pub enum SyncCommand { | |
| ChangeSyncPoint { slot: Slot, hash: BlockHash }, | |
| } | |
| pub enum ChainSyncCommand { | |
| FindIntersect { slot: Slot, hash: BlockHash }, | |
| } |
Not a huge deal, but IMO "sync" is a bit of a generic name when we're interacting with a specific miniprotocol
There was a problem hiding this comment.
Renamed to ChainSyncCommand in 76cd5fb.
modules/indexer/src/indexer.rs
Outdated
| // Configuration defaults | ||
| const DEFAULT_DYNAMIC_SYNC_TOPIC: (&str, &str) = | ||
| ("dynamic-sync-publisher-topic", "cardano.sync.command"); | ||
|
|
There was a problem hiding this comment.
Can we follow the strongly-typed configuration pattern which PeerNetworkInterface is using? I'd like that pattern to infect as much of the system as possible.
There was a problem hiding this comment.
Updated to use the same pattern as PeerNetworkInterface in 3dfdb78.
| Ok(Point::Origin) => { | ||
| warn!("Dynamic sync received Point::Origin; ignoring"); | ||
| return; | ||
| } |
There was a problem hiding this comment.
I don't think there's any reason to not support syncing to origin.
| SyncPoint::Dynamic => { | ||
| let mut rx = match cmd_rx.take() { | ||
| Some(rx) => rx, | ||
| None => { | ||
| warn!("Dynamic mode configured but cmd_rx is missing"); | ||
| return; | ||
| } | ||
| }; | ||
|
|
||
| let point = match Self::wait_sync_command(&mut rx).await { | ||
| Ok(Point::Specific(slot, hash)) => { | ||
| let (epoch, _) = sink.genesis_values.slot_to_epoch(slot); | ||
| sink.last_epoch = Some(epoch); | ||
| info!("Dynamic sync starting at slot {} (epoch {})", slot, epoch); | ||
| Point::Specific(slot, hash) | ||
| } | ||
| Ok(Point::Origin) => { | ||
| warn!("Dynamic sync received Point::Origin; ignoring"); | ||
| return; | ||
| } | ||
| Err(err) => { | ||
| warn!("Failed to receive initial sync command: {err:#}"); | ||
| return; | ||
| } | ||
| }; | ||
|
|
||
| let mut manager = Self::init_manager(cfg, sink, Some(rx)); | ||
| manager.sync_to_point(point); | ||
| manager | ||
| } |
There was a problem hiding this comment.
| SyncPoint::Dynamic => { | |
| let mut rx = match cmd_rx.take() { | |
| Some(rx) => rx, | |
| None => { | |
| warn!("Dynamic mode configured but cmd_rx is missing"); | |
| return; | |
| } | |
| }; | |
| let point = match Self::wait_sync_command(&mut rx).await { | |
| Ok(Point::Specific(slot, hash)) => { | |
| let (epoch, _) = sink.genesis_values.slot_to_epoch(slot); | |
| sink.last_epoch = Some(epoch); | |
| info!("Dynamic sync starting at slot {} (epoch {})", slot, epoch); | |
| Point::Specific(slot, hash) | |
| } | |
| Ok(Point::Origin) => { | |
| warn!("Dynamic sync received Point::Origin; ignoring"); | |
| return; | |
| } | |
| Err(err) => { | |
| warn!("Failed to receive initial sync command: {err:#}"); | |
| return; | |
| } | |
| }; | |
| let mut manager = Self::init_manager(cfg, sink, Some(rx)); | |
| manager.sync_to_point(point); | |
| manager | |
| } | |
| SyncPoint::Dynamic => { | |
| let manager = Self::init_manager(cfg, sink, Some(rx)); | |
| manager | |
| } |
If we can handle these commands while the manager is running, I don't think we need to handle them before it starts
There was a problem hiding this comment.
I've implemented this feedback in d9110c1.
| // Create background task to foward sync commands to NetworkManager | ||
| let mut cmd_rx = if cfg.sync_point == SyncPoint::Dynamic { | ||
| Some(Self::spawn_command_forwarder(context.clone(), &cfg.sync_command_topic).await?) | ||
| } else { | ||
| None | ||
| }; |
There was a problem hiding this comment.
Having a dedicated channel receiving sync commands works, but I think that these commands should be sent to NetworkManager by writing to its event queue instead.
-
Future requirements: there are going to be more commands in the future, the consensus module will be able to tell this module to follow a different chain/peer and I suspect someday a "peer discovery" module will tell it to connect to new peers or disconnect from shady ones. So this module will have to handle several kinds of commands from outside, and that'll be less onerous if we do it by publishing them all to one queue.
-
I don't think we need to restrict "sync commands" so that they only work when you're using this "dynamic" sync point. We can change which point we sync to at runtime, which means that we could start by syncing to the tip and switch to something else later. I don't have a concrete use case in mind for that, but in general I try not to block functionality if it's easy to support.
There was a problem hiding this comment.
Refactored the module to follow this approach in d9110c1. Sync commands are now forwarded into the NetworkManager through the same event queue used for peer updates.
Signed-off-by: William Hankins <[email protected]>
Signed-off-by: William Hankins <[email protected]>
Signed-off-by: William Hankins <[email protected]>
Signed-off-by: William Hankins <[email protected]>
Signed-off-by: William Hankins <[email protected]>
Description
This PR introduces a new
dynamicsync mode to thepeer_network_interface. In this mode, the interface waits for aChangeSyncPointcommand on thecardano.sync.commandtopic and begins fetching from that point.A background task forwards incoming sync points to the
NetworkManager. When a new point arrives, the manager resets theChainState, clears outstanding peer requests, and issues a newfind_intersectacross all upstream peers.A minimal
indexerprocess is included to demonstrate this behavior. It registers only the required modules and emits twoChangeSyncPointcommands 5 seconds apart, demonstrating dynamic re-syncing.Related Issue(s)
#374
How was this tested?
By running the
indexerprocess:peer_network_interfacebegins fetching blocks normally.ChangeSyncPointis published, returning to the starting point.peer_network_interfaceresets and syncs again from that point.For visibility, you can log inside
handle_roll_forwardto observe block fetching.Checklist
Impact / Side effects
Adds a basic
indexerprocess intended to become the minimal working example for building custom indexers on top of Acropolis.Reviewer notes / Areas to focus
NetworkManagerconstruction inpeer_network_interface/src/peer_network_interface.rs.on_sync_cmdwhen resetting sync state inpeer_network_interface/src/network.rs.