mab122
/
radicle-heartwood-lfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

node: Handle `fetched` differently if initiated

This commit is contained in:
Alexis Sellier 2023-01-30 13:22:33 +01:00
parent 3aa105afda
commit 498a5c4e4a
No known key found for this signature in database
4 changed files with 48 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
radicle-node/src/service.rs
View File

@ -512,53 +512,56 @@ where
let remote = result.remote;
let rid = result.rid;
let namespaces = result.namespaces;
let initiated = result.initiated;
log::debug!(
target: "service",
"Fetched {rid} {remote} (error={:?})", result.result.as_ref().err()
);
let result = match result.result {
Ok(updated) => {
self.reactor.event(Event::RefsFetched {
remote,
rid,
updated: updated.clone(),
});
Ok(updated)
}
Err(err) => {
error!(target: "service", "Fetch failed for {rid} from {remote}: {err}");
if let FetchError::Io(_) = err {
self.reactor
.disconnect(result.remote, DisconnectReason::Fetch(err));
return;
} else {
Err(err)
if initiated {
log::debug!(
target: "service",
"Fetched {rid} {remote} (error={:?})", result.result.as_ref().err()
);
let result = match result.result {
Ok(updated) => {
self.reactor.event(Event::RefsFetched {
remote,
rid,
updated: updated.clone(),
});
Ok(updated)
}
}
};
Err(err) => {
error!(target: "service", "Fetch failed for {rid} from {remote}: {err}");
if let Some(results) = self.fetch_reqs.get(&rid) {
log::debug!(target: "service", "Found existing fetch request, sending result..");
if let FetchError::Io(_) = err {
self.reactor
.disconnect(result.remote, DisconnectReason::Fetch(err));
return;
} else {
Err(err)
}
}
};
if results
.send(FetchResult {
rid,
remote,
namespaces,
result,
})
.is_err()
{
log::error!(target: "service", "Error sending fetch result for {rid}..");
self.fetch_reqs.remove(&rid);
if let Some(results) = self.fetch_reqs.get(&rid) {
log::debug!(target: "service", "Found existing fetch request, sending result..");
if results
.send(FetchResult {
rid,
initiated,
remote,
namespaces,
result,
})
.is_err()
{
log::error!(target: "service", "Error sending fetch result for {rid}..");
self.fetch_reqs.remove(&rid);
} else {
log::debug!(target: "service", "Sent fetch result for {rid}..");
}
} else {
log::debug!(target: "service", "Sent fetch result for {rid}..");
log::debug!(target: "service", "No fetch requests found for {rid}..");
}
} else {
log::debug!(target: "service", "No fetch requests found for {rid}..");
}
if let Some(session) = self.sessions.get_mut(&remote) {

2
radicle-node/src/test/simulator.rs
View File

@ -621,6 +621,7 @@ impl<S: WriteStorage + 'static, G: Signer> Simulation<S, G> {
remote: fetch.remote,
input: Input::Fetched(Arc::new(FetchResult {
rid: fetch.repo,
initiated: fetch.initiated,
remote: fetch.remote,
namespaces: fetch.namespaces,
result: Err(FetchError::Io(io::ErrorKind::Other.into())),
@ -635,6 +636,7 @@ impl<S: WriteStorage + 'static, G: Signer> Simulation<S, G> {
remote: fetch.remote,
input: Input::Fetched(Arc::new(FetchResult {
rid: fetch.repo,
initiated: fetch.initiated,
remote: fetch.remote,
namespaces: fetch.namespaces,
result: Ok(vec![]),

1
radicle-node/src/worker.rs
View File

@ -81,6 +81,7 @@ impl<G: Signer + EcSign + 'static> Worker<G> {
rid: fetch.repo,
remote: fetch.remote,
namespaces: fetch.namespaces,
initiated: fetch.initiated,
result,
};
log::debug!(target: "worker", "Sending response back to service..");

1
radicle/src/node.rs
View File

@ -61,6 +61,7 @@ impl From<net::SocketAddr> for Address {
pub struct FetchResult {
pub rid: Id,
pub remote: NodeId,
pub initiated: bool,
pub namespaces: Namespaces,
pub result: Result<Vec<RefUpdate>, FetchError>,
}