node: Handle refs update announcement
Signed-off-by: Alexis Sellier <self@cloudhead.io>
This commit is contained in:
parent
ff2bd185e7
commit
14395ad873
|
|
@ -56,7 +56,7 @@ pub struct Handle<R: Reactor> {
|
|||
impl<R: Reactor> traits::Handle for Handle<R> {
|
||||
/// Notify the client that a project has been updated.
|
||||
fn updated(&self, id: ProjId) -> Result<(), Error> {
|
||||
self.command(protocol::Command::AnnounceInventory(id))
|
||||
self.command(protocol::Command::AnnounceRefsUpdate(id))
|
||||
}
|
||||
|
||||
/// Send a command to the command channel, and wake up the event loop.
|
||||
|
|
|
|||
|
|
@ -1,13 +1,15 @@
|
|||
use std::str::FromStr;
|
||||
|
||||
use git_ref_format as format;
|
||||
use git_url::Url;
|
||||
use radicle_git_ext as git_ext;
|
||||
|
||||
use crate::collections::HashMap;
|
||||
use crate::identity::UserId;
|
||||
use crate::storage::{Remote, Remotes, Unverified};
|
||||
|
||||
pub use git_ext::Oid;
|
||||
pub use git_url::Url;
|
||||
|
||||
/// Default port of the `git` transport protocol.
|
||||
pub const PROTOCOL_PORT: u16 = 9418;
|
||||
|
||||
|
|
|
|||
|
|
@ -19,11 +19,11 @@ use crate::address_manager::AddressManager;
|
|||
use crate::clock::RefClock;
|
||||
use crate::collections::{HashMap, HashSet};
|
||||
use crate::crypto;
|
||||
use crate::identity::ProjId;
|
||||
use crate::identity::{ProjId, UserId};
|
||||
use crate::protocol::config::ProjectTracking;
|
||||
use crate::protocol::message::Message;
|
||||
use crate::protocol::peer::{Peer, PeerError, PeerState};
|
||||
use crate::storage::{self, WriteRepository};
|
||||
use crate::storage::{self, ReadRepository, WriteRepository};
|
||||
use crate::storage::{Inventory, ReadStorage, Remotes, Unverified, WriteStorage};
|
||||
|
||||
pub use crate::protocol::config::{Config, Network};
|
||||
|
|
@ -54,7 +54,7 @@ pub enum Event {}
|
|||
pub enum Command {
|
||||
Connect(net::SocketAddr),
|
||||
Fetch(ProjId, net::SocketAddr),
|
||||
AnnounceInventory(ProjId),
|
||||
AnnounceRefsUpdate(ProjId),
|
||||
}
|
||||
|
||||
/// Command-related errors.
|
||||
|
|
@ -335,11 +335,25 @@ where
|
|||
})
|
||||
.unwrap();
|
||||
}
|
||||
Command::AnnounceInventory(proj) => {
|
||||
Command::AnnounceRefsUpdate(proj) => {
|
||||
let user = *self.storage.user_id();
|
||||
let repo = self.storage.repository(&proj).unwrap();
|
||||
let remote = repo.remote(&user).unwrap();
|
||||
let refs = remote.refs;
|
||||
let signature = self
|
||||
.signer
|
||||
.sign(serde_json::to_vec(&refs).unwrap().as_slice());
|
||||
let peers = self.peers.negotiated().map(|(_, p)| p.addr);
|
||||
|
||||
self.context
|
||||
.broadcast(Message::InventoryUpdate { inv: vec![proj] }, peers);
|
||||
self.context.broadcast(
|
||||
Message::RefsUpdate {
|
||||
proj,
|
||||
user,
|
||||
refs,
|
||||
signature,
|
||||
},
|
||||
peers,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -613,12 +627,13 @@ where
|
|||
}
|
||||
|
||||
/// Process a peer inventory update announcement by (maybe) fetching.
|
||||
fn process_inventory_update(&mut self, inventory: &Inventory, _from: NodeId, remote: &Url) {
|
||||
for proj_id in inventory {
|
||||
if self.config.is_tracking(proj_id) {
|
||||
self.fetch(proj_id, remote);
|
||||
}
|
||||
fn process_refs_update(&mut self, proj: &ProjId, _user: &UserId, remote: &Url) -> bool {
|
||||
// TODO: Check that we're tracking this user as well.
|
||||
if self.config.is_tracking(proj) {
|
||||
self.fetch(proj, remote);
|
||||
}
|
||||
// TODO: If refs were updated, return `true`.
|
||||
false
|
||||
}
|
||||
|
||||
fn fetch(&mut self, proj_id: &ProjId, remote: &Url) {
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ use git_url::Url;
|
|||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::crypto;
|
||||
use crate::identity::ProjId;
|
||||
use crate::identity::{ProjId, UserId};
|
||||
use crate::protocol::{Context, NodeId, Timestamp, PROTOCOL_VERSION};
|
||||
use crate::storage;
|
||||
|
||||
|
|
@ -86,9 +86,7 @@ pub enum Message {
|
|||
announcement: NodeAnnouncement,
|
||||
},
|
||||
/// Get a peer's inventory.
|
||||
GetInventory {
|
||||
ids: Vec<ProjId>,
|
||||
},
|
||||
GetInventory { ids: Vec<ProjId> },
|
||||
/// Send our inventory to a peer. Sent in response to [`Message::GetInventory`].
|
||||
/// Nb. This should be the whole inventory, not a partial update.
|
||||
Inventory {
|
||||
|
|
@ -98,8 +96,17 @@ pub enum Message {
|
|||
/// are the originator, only when relaying.
|
||||
origin: Option<NodeId>,
|
||||
},
|
||||
InventoryUpdate {
|
||||
inv: Vec<ProjId>,
|
||||
/// Project refs were updated. Includes the signature of the user who updated
|
||||
/// their view of the project.
|
||||
RefsUpdate {
|
||||
/// Project under which the refs were updated.
|
||||
proj: ProjId,
|
||||
/// User signing.
|
||||
user: UserId,
|
||||
/// Updated refs.
|
||||
refs: storage::Refs,
|
||||
/// Signature over the refs.
|
||||
signature: crypto::Signature,
|
||||
},
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -194,9 +194,28 @@ impl Peer {
|
|||
}));
|
||||
}
|
||||
}
|
||||
(PeerState::Negotiated { id, git, .. }, Message::InventoryUpdate { inv }) => {
|
||||
// TODO: Buffer/throttle fetches.
|
||||
ctx.process_inventory_update(&inv, *id, git);
|
||||
(
|
||||
PeerState::Negotiated { git, .. },
|
||||
Message::RefsUpdate {
|
||||
proj,
|
||||
user,
|
||||
refs,
|
||||
signature,
|
||||
},
|
||||
) => {
|
||||
let bytes = serde_json::to_vec(&refs).unwrap();
|
||||
|
||||
if user.verify(&signature, &bytes).is_ok() {
|
||||
// TODO: Buffer/throttle fetches.
|
||||
// TODO: Also pass the updated refs so that we can check whether
|
||||
// we need to fetch or not.
|
||||
// TODO: Check that the refs are valid.
|
||||
if ctx.process_refs_update(&proj, &user, git) {
|
||||
// TODO: If refs were updated, propagate message to peers.
|
||||
}
|
||||
} else {
|
||||
return Err(PeerError::Misbehavior);
|
||||
}
|
||||
}
|
||||
(
|
||||
PeerState::Negotiated { .. },
|
||||
|
|
|
|||
|
|
@ -42,7 +42,7 @@ pub enum Error {
|
|||
InvalidHead,
|
||||
}
|
||||
|
||||
pub type Refs = HashMap<BranchName, Oid>;
|
||||
pub type Refs = HashMap<RefName, Oid>;
|
||||
pub type RemoteId = UserId;
|
||||
pub type RefName = String;
|
||||
|
||||
|
|
@ -130,6 +130,7 @@ pub trait WriteStorage: ReadStorage {
|
|||
|
||||
pub trait ReadRepository {
|
||||
fn path(&self) -> &Path;
|
||||
fn remote(&self, user: &UserId) -> Result<Remote<Unverified>, Error>;
|
||||
fn remotes(&self) -> Result<Remotes<Unverified>, Error>;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -144,6 +144,24 @@ impl ReadRepository for Repository {
|
|||
self.backend.path()
|
||||
}
|
||||
|
||||
fn remote(&self, user: &UserId) -> Result<Remote<Unverified>, Error> {
|
||||
// TODO: Only fetch standard refs.
|
||||
let entries = self
|
||||
.backend
|
||||
.references_glob(format!("refs/remotes/{}/*", user).as_str())?;
|
||||
let mut refs = HashMap::default();
|
||||
|
||||
for e in entries {
|
||||
let e = e?;
|
||||
let name = e.name().ok_or(Error::InvalidRef)?;
|
||||
let (_, refname) = git::parse_ref::<UserId>(name)?;
|
||||
let oid = e.target().ok_or(Error::InvalidRef)?;
|
||||
|
||||
refs.insert(refname.to_string(), oid.into());
|
||||
}
|
||||
Ok(Remote::new(*user, refs))
|
||||
}
|
||||
|
||||
fn remotes(&self) -> Result<Remotes<Unverified>, Error> {
|
||||
let refs = self.backend.references_glob(NAMESPACES_GLOB.as_str())?;
|
||||
let mut remotes = HashMap::default();
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ use git_url::Url;
|
|||
|
||||
use crate::identity::{ProjId, UserId};
|
||||
use crate::storage::{
|
||||
Error, Inventory, ReadRepository, ReadStorage, Remotes, Unverified, WriteRepository,
|
||||
Error, Inventory, ReadRepository, ReadStorage, Remote, Remotes, Unverified, WriteRepository,
|
||||
WriteStorage,
|
||||
};
|
||||
|
||||
|
|
@ -69,6 +69,10 @@ impl ReadRepository for MockRepository {
|
|||
todo!()
|
||||
}
|
||||
|
||||
fn remote(&self, _user: &UserId) -> Result<Remote<Unverified>, Error> {
|
||||
todo!()
|
||||
}
|
||||
|
||||
fn remotes(&self) -> Result<Remotes<Unverified>, Error> {
|
||||
todo!()
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue