smeb_y/0g-storage-node
1
0
Fork 0
mirror of https://github.com/0glabs/0g-storage-node.git synced 2025-04-04 15:35:18 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Publish NewFile message when any file finalized

Browse Source
This commit is contained in:
boqiu 2024-10-23 17:05:55 +08:00
parent e06936f381
commit d12bf66129
3 changed files with 22 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
node/router/src/service.rs
View File

@ -5,11 +5,14 @@ use chunk_pool::ChunkPoolMessage;
use file_location_cache::FileLocationCache; use file_location_cache::FileLocationCache;
use futures::{channel::mpsc::Sender, prelude::*}; use futures::{channel::mpsc::Sender, prelude::*};
use miner::MinerMessage; use miner::MinerMessage;
use network::types::NewFile;
use network::PubsubMessage;
use network::{ use network::{
BehaviourEvent, Keypair, Libp2pEvent, NetworkGlobals, NetworkMessage, RequestId, BehaviourEvent, Keypair, Libp2pEvent, NetworkGlobals, NetworkMessage, RequestId,
Service as LibP2PService, Swarm, Service as LibP2PService, Swarm,
}; };
use pruner::PrunerMessage; use pruner::PrunerMessage;
use shared_types::timestamp_now;
use std::sync::Arc; use std::sync::Arc;
use storage::log_store::Store as LogStore; use storage::log_store::Store as LogStore;
use storage_async::Store; use storage_async::Store;
@ -44,6 +47,8 @@ pub struct RouterService {
/// Stores potentially created UPnP mappings to be removed on shutdown. (TCP port and UDP /// Stores potentially created UPnP mappings to be removed on shutdown. (TCP port and UDP
/// port). /// port).
upnp_mappings: (Option<u16>, Option<u16>), upnp_mappings: (Option<u16>, Option<u16>),
store: Arc<dyn LogStore>,
} }
impl RouterService { impl RouterService {
@ -63,7 +68,6 @@ impl RouterService {
local_keypair: Keypair, local_keypair: Keypair,
config: Config, config: Config,
) { ) {
let store = Store::new(store, executor.clone());
let peers = Arc::new(RwLock::new(PeerManager::new(config.clone()))); let peers = Arc::new(RwLock::new(PeerManager::new(config.clone())));
// create the network service and spawn the task // create the network service and spawn the task
@ -81,11 +85,12 @@ impl RouterService {
sync_send, sync_send,
chunk_pool_send, chunk_pool_send,
local_keypair, local_keypair,
store, Store::new(store.clone(), executor.clone()),
file_location_cache, file_location_cache,
peers, peers,
), ),
upnp_mappings: (None, None), upnp_mappings: (None, None),
store,
}; };
// spawn service // spawn service
@ -328,15 +333,16 @@ impl RouterService {
} }
} }
NetworkMessage::AnnounceLocalFile { tx_id } => { NetworkMessage::AnnounceLocalFile { tx_id } => {
if self let shard_config = self.store.get_shard_config();
.libp2p_event_handler let msg = PubsubMessage::NewFile(NewFile {
.publish_file(tx_id) tx_id,
.await num_shard: shard_config.num_shard,
.is_some() shard_id: shard_config.shard_id,
{ timestamp: timestamp_now(),
});
self.libp2p.swarm.behaviour_mut().publish(vec![msg]);
metrics::SERVICE_ROUTE_NETWORK_MESSAGE_ANNOUNCE_LOCAL_FILE.mark(1); metrics::SERVICE_ROUTE_NETWORK_MESSAGE_ANNOUNCE_LOCAL_FILE.mark(1);
} }
}
NetworkMessage::UPnPMappingEstablished { NetworkMessage::UPnPMappingEstablished {
tcp_socket, tcp_socket,
udp_socket, udp_socket,

2
node/sync/src/controllers/serial.rs
View File

@ -545,6 +545,8 @@ impl SerialSyncController {
info!(%self.tx_seq, "Succeeded to finalize file"); info!(%self.tx_seq, "Succeeded to finalize file");
self.state = SyncState::Completed; self.state = SyncState::Completed;
metrics::SERIAL_SYNC_FILE_COMPLETED.update_since(self.since.0); metrics::SERIAL_SYNC_FILE_COMPLETED.update_since(self.since.0);
self.ctx
.send(NetworkMessage::AnnounceLocalFile { tx_id: self.tx_id });
} }
Ok(false) => { Ok(false) => {
warn!(?self.tx_id, %self.tx_seq, "Transaction reverted during finalize_tx"); warn!(?self.tx_id, %self.tx_seq, "Transaction reverted during finalize_tx");

5
node/sync/src/service.rs
View File

@ -642,7 +642,10 @@ impl SyncService {
Some(s) => s, Some(s) => s,
None => { None => {
debug!(%tx.seq, "No more data needed"); debug!(%tx.seq, "No more data needed");
self.store.finalize_tx_with_hash(tx.seq, tx.hash()).await?; if self.store.finalize_tx_with_hash(tx.seq, tx.hash()).await? {
self.ctx
.send(NetworkMessage::AnnounceLocalFile { tx_id: tx.id() });
}
return Ok(()); return Ok(());
} }
}; };