1use std::collections::{BTreeSet, HashMap, VecDeque};
2use std::io::{self, BufRead, Write};
3use std::path::PathBuf;
4use std::sync::{Arc, Condvar, Mutex};
5use std::time::{Duration, Instant};
6
7use rns_core::msgpack::{self, Value};
8use rns_core::types::{DestHash, LinkId};
9use rns_net::{AnnouncedIdentity, Callbacks, PacketHash, RnsNode};
10
11use crate::config::ClientConfig;
12use crate::logging;
13use crate::protocol::{self, RefUpdate};
14use crate::util::{
15 default_reticulum_dir, default_rngit_dir, load_or_create_identity, parse_rns_url,
16};
17use crate::{git, Error, Result};
18
19pub fn main<I>(args: I) -> Result<()>
20where
21 I: IntoIterator<Item = String>,
22{
23 let options = ClientOptions::parse(args)?;
24 let (dest_hash, repository) = parse_rns_url(&options.url)?;
25 let rngit_dir = options.config_dir.unwrap_or_else(default_rngit_dir);
26 let rns_dir = options.rns_config_dir.or_else(default_reticulum_dir);
27 let (config, created) = ClientConfig::load_or_create(rngit_dir, rns_dir)?;
28 logging::init_file_logger(&config.dir.join("client_log"), config.log_level)?;
29 if created {
30 return Err(Error::msg(format!(
31 "created default config at {}; edit it and run again",
32 config.dir.join("client_config").display()
33 )));
34 }
35
36 let helper = RemoteHelper::connect(config, dest_hash)?;
37 helper.run(repository)
38}
39
40struct RemoteHelper {
41 client: SyncClient,
42 remote_refs: Mutex<HashMap<String, String>>,
43}
44
45impl RemoteHelper {
46 fn connect(config: ClientConfig, dest_hash: [u8; 16]) -> Result<Self> {
47 Ok(Self {
48 client: SyncClient::connect(config, dest_hash)?,
49 remote_refs: Mutex::new(HashMap::new()),
50 })
51 }
52
53 fn run(self, repository: String) -> Result<()> {
54 let stdin = io::stdin();
55 let stdout = io::stdout();
56 self.run_io(repository, stdin.lock(), stdout.lock())
57 }
58
59 fn run_io<R: BufRead, W: Write>(
60 &self,
61 repository: String,
62 mut input: R,
63 mut output: W,
64 ) -> Result<()> {
65 let mut line = String::new();
66 let mut fetch_refs = Vec::new();
67 let mut push_specs = Vec::new();
68
69 loop {
70 line.clear();
71 if input.read_line(&mut line)? == 0 {
72 break;
73 }
74 let command = line.trim_end();
75 if command.is_empty() {
76 if !fetch_refs.is_empty() {
77 self.fetch(&repository, &fetch_refs)?;
78 fetch_refs.clear();
79 writeln!(output)?;
80 output.flush()?;
81 }
82 if !push_specs.is_empty() {
83 for spec in push_specs.drain(..) {
84 self.push(&repository, &spec)?;
85 writeln!(output, "ok {}", spec.remote)?;
86 }
87 writeln!(output)?;
88 output.flush()?;
89 }
90 continue;
91 }
92
93 if command == "capabilities" {
94 writeln!(output, "option")?;
95 writeln!(output, "list")?;
96 writeln!(output, "fetch")?;
97 writeln!(output, "push")?;
98 writeln!(output)?;
99 } else if command == "list" || command == "list for-push" {
100 let refs = self.list(&repository)?;
101 output.write_all(&refs)?;
102 writeln!(output)?;
103 } else if let Some(rest) = command.strip_prefix("option ") {
104 let _ = rest;
105 writeln!(output, "ok")?;
106 } else if let Some(rest) = command.strip_prefix("fetch ") {
107 fetch_refs.push(parse_fetch_command(rest)?);
108 } else if let Some(rest) = command.strip_prefix("push ") {
109 push_specs.push(parse_push_spec(rest)?);
110 } else {
111 writeln!(output, "error unsupported command")?;
112 }
113 output.flush()?;
114 }
115 Ok(())
116 }
117
118 fn list(&self, repository: &str) -> Result<Vec<u8>> {
119 let response = self.client.request(
120 protocol::PATH_LIST,
121 protocol::repository_request(repository),
122 )?;
123 let bytes = protocol::response_bin(&response.data)?;
124 let refs = decode_status(bytes)?;
125 self.record_remote_refs(&refs);
126 Ok(refs)
127 }
128
129 fn fetch(&self, repository: &str, refs: &[FetchRef]) -> Result<()> {
130 let have = self.fetch_have_set(refs);
131 let response = self.client.request(
132 protocol::PATH_FETCH,
133 protocol::fetch_request(repository, &have),
134 )?;
135 if let Some(metadata) = response.metadata {
136 ensure_metadata_ok(&metadata)?;
137 let bundle = protocol::response_bin(&response.data)?;
138 git::fetch_bundle_into_local(
139 &bundle,
140 &refs.iter().map(|r| r.name.clone()).collect::<Vec<_>>(),
141 )?;
142 return Ok(());
143 }
144 let bytes = protocol::response_bin(&response.data)?;
145 let bundle = decode_status(bytes)?;
146 if !bundle.is_empty() {
147 git::fetch_bundle_into_local(
148 &bundle,
149 &refs.iter().map(|r| r.name.clone()).collect::<Vec<_>>(),
150 )?;
151 }
152 Ok(())
153 }
154
155 fn push(&self, repository: &str, spec: &PushSpec) -> Result<()> {
156 let mut updates = Vec::new();
157 let mut bundle_refs = Vec::new();
158 if spec.local.is_empty() {
159 updates.push(RefUpdate {
160 refname: spec.remote.clone(),
161 old: None,
162 new: None,
163 force: spec.force,
164 });
165 } else {
166 let sha = git::local_ref_sha(&spec.local)?
167 .ok_or_else(|| Error::msg(format!("unknown local ref {}", spec.local)))?;
168 bundle_refs.push(spec.local.clone());
169 updates.push(RefUpdate {
170 refname: spec.remote.clone(),
171 old: None,
172 new: Some(sha),
173 force: spec.force,
174 });
175 }
176
177 let exclusions = self.push_exclusion_set();
178 let bundle = git::create_local_bundle(&bundle_refs, &exclusions)?;
179 let response = self.client.request(
180 protocol::PATH_PUSH,
181 protocol::push_request(repository, bundle, updates),
182 )?;
183 let bytes = protocol::response_bin(&response.data)?;
184 let _ = decode_status(bytes)?;
185 Ok(())
186 }
187
188 fn record_remote_refs(&self, refs: &[u8]) {
189 *self.remote_refs.lock().unwrap() = parse_remote_refs(refs);
190 }
191
192 fn fetch_have_set(&self, refs: &[FetchRef]) -> Vec<String> {
193 let remote_refs = self.remote_refs.lock().unwrap();
194 build_fetch_have_set(&remote_refs, refs, |refname| {
195 git::local_ref_sha(refname).ok().flatten()
196 })
197 }
198
199 fn push_exclusion_set(&self) -> Vec<String> {
200 let remote_refs = self.remote_refs.lock().unwrap();
201 build_push_exclusion_set(&remote_refs, git::object_exists_local)
202 }
203}
204
205fn parse_remote_refs(refs: &[u8]) -> HashMap<String, String> {
206 let Ok(text) = std::str::from_utf8(refs) else {
207 return HashMap::new();
208 };
209 text.lines()
210 .filter_map(|line| {
211 let (sha, name) = line.split_once(' ')?;
212 (name != "HEAD").then(|| (name.to_string(), sha.to_string()))
213 })
214 .collect()
215}
216
217fn build_fetch_have_set(
218 remote_refs: &HashMap<String, String>,
219 refs: &[FetchRef],
220 resolve_local_ref: impl Fn(&str) -> Option<String>,
221) -> Vec<String> {
222 let mut have = BTreeSet::new();
223 for fetch_ref in refs {
224 if let Some(local_sha) = resolve_local_ref(&fetch_ref.name) {
225 if local_sha != fetch_ref.sha {
226 have.insert(local_sha);
227 }
228 }
229 }
230 for (refname, remote_sha) in remote_refs {
231 if let Some(local_sha) = resolve_local_ref(refname) {
232 if &local_sha == remote_sha {
233 have.insert(local_sha);
234 }
235 }
236 }
237 have.into_iter().collect()
238}
239
240fn build_push_exclusion_set(
241 remote_refs: &HashMap<String, String>,
242 object_exists: impl Fn(&str) -> bool,
243) -> Vec<String> {
244 remote_refs
245 .values()
246 .filter(|sha| object_exists(sha))
247 .cloned()
248 .collect::<BTreeSet<_>>()
249 .into_iter()
250 .collect()
251}
252
253pub(crate) struct SyncClient {
254 node: RnsNode,
255 link_id: [u8; 16],
256 state: Arc<(Mutex<ClientState>, Condvar)>,
257 request_timeout: Duration,
258}
259
260impl SyncClient {
261 pub(crate) fn connect(config: ClientConfig, dest_hash: [u8; 16]) -> Result<Self> {
262 let callbacks = SharedCallbacks::default();
263 let state = callbacks.state.clone();
264 let node = RnsNode::from_config(config.reticulum_dir.as_deref(), Box::new(callbacks))?;
265 let client_identity = load_or_create_identity(&config.identity_path)?;
266
267 let dest = DestHash(dest_hash);
268 eprintln!("Requesting path to {}...", crate::util::hex(&dest_hash));
269 node.request_path(&dest)
270 .map_err(|_| Error::msg("failed to request destination path"))?;
271 let deadline = Instant::now() + Duration::from_secs(config.connect_timeout_secs);
272 while !node.has_path(&dest).unwrap_or(false) && Instant::now() < deadline {
273 std::thread::sleep(Duration::from_millis(250));
274 }
275 if !node.has_path(&dest).unwrap_or(false) {
276 return Err(Error::msg("destination path resolution timed out"));
277 }
278 eprintln!("Path resolved.");
279
280 let recalled = node
281 .recall_identity(&dest)
282 .map_err(|_| Error::msg("failed to recall destination identity"))?
283 .ok_or_else(|| Error::msg("destination identity is unknown"))?;
284 let sig_pub: [u8; 32] = recalled.public_key[32..64].try_into().unwrap();
285 let link_id = node
286 .create_link(dest_hash, sig_pub)
287 .map_err(|_| Error::msg("failed to create RNS link"))?;
288 eprintln!("Establishing link...");
289 let private_key = client_identity
290 .get_private_key()
291 .ok_or_else(|| Error::msg("client identity has no private key"))?;
292 node.identify_on_link(link_id, private_key)
293 .map_err(|_| Error::msg("failed to identify on RNS link"))?;
294
295 wait_for_link(
296 &state,
297 link_id,
298 Duration::from_secs(config.connect_timeout_secs),
299 )?;
300 eprintln!("Link established.");
301 Ok(Self {
302 node,
303 link_id,
304 state,
305 request_timeout: Duration::from_secs(config.request_timeout_secs),
306 })
307 }
308
309 pub(crate) fn request(&self, path: &str, data: Vec<u8>) -> Result<Response> {
310 {
311 let (lock, _) = &*self.state;
312 lock.lock().unwrap().responses.clear();
313 }
314 self.node
315 .send_request(self.link_id, path, &data)
316 .map_err(|_| Error::msg("failed to send request"))?;
317 let deadline = Instant::now() + self.request_timeout;
318 let (lock, cv) = &*self.state;
319 let mut state = lock.lock().unwrap();
320 loop {
321 if let Some(response) = state.responses.pop_front() {
322 return Ok(response);
323 }
324 let now = Instant::now();
325 if now >= deadline {
326 return Err(Error::msg("request timed out"));
327 }
328 let wait = deadline.saturating_duration_since(now);
329 let (next, _) = cv.wait_timeout(state, wait).unwrap();
330 state = next;
331 }
332 }
333}
334
335#[derive(Default)]
336struct SharedCallbacks {
337 state: Arc<(Mutex<ClientState>, Condvar)>,
338}
339
340#[derive(Default)]
341struct ClientState {
342 established: Vec<[u8; 16]>,
343 responses: VecDeque<Response>,
344 progress: ProgressState,
345}
346
347#[derive(Default)]
348struct ProgressState {
349 started_at: Option<Instant>,
350 last_percent: Option<usize>,
351}
352
353#[derive(Debug, Clone)]
354pub(crate) struct Response {
355 pub(crate) data: Vec<u8>,
356 pub(crate) metadata: Option<Vec<u8>>,
357}
358
359impl Callbacks for SharedCallbacks {
360 fn on_announce(&mut self, _announced: AnnouncedIdentity) {}
361
362 fn on_path_updated(&mut self, _dest_hash: DestHash, _hops: u8) {}
363
364 fn on_local_delivery(&mut self, _dest_hash: DestHash, _raw: Vec<u8>, _packet_hash: PacketHash) {
365 }
366
367 fn on_link_established(
368 &mut self,
369 link_id: LinkId,
370 _dest_hash: DestHash,
371 _rtt: f64,
372 _is_initiator: bool,
373 ) {
374 let (lock, cv) = &*self.state;
375 lock.lock().unwrap().established.push(link_id.0);
376 cv.notify_all();
377 }
378
379 fn on_response_with_metadata(
380 &mut self,
381 _link_id: LinkId,
382 _request_id: [u8; 16],
383 data: Vec<u8>,
384 metadata: Option<Vec<u8>>,
385 ) {
386 let (lock, cv) = &*self.state;
387 lock.lock()
388 .unwrap()
389 .responses
390 .push_back(Response { data, metadata });
391 cv.notify_all();
392 }
393
394 fn on_resource_progress(&mut self, _link_id: LinkId, received: usize, total: usize) {
395 if total == 0 {
396 return;
397 }
398 let (lock, _) = &*self.state;
399 let mut state = lock.lock().unwrap();
400 let progress = &mut state.progress;
401 let started_at = *progress.started_at.get_or_insert_with(Instant::now);
402 let percent = ((received.saturating_mul(100)) / total).min(100);
403 if progress.last_percent == Some(percent) && received < total {
404 return;
405 }
406 progress.last_percent = Some(percent);
407 let elapsed = started_at.elapsed().as_secs_f64().max(0.001);
408 let rate = received as f64 / elapsed;
409 eprintln!("rns: transfer {percent}% ({received}/{total} parts, {rate:.1} parts/s)");
410 if received >= total {
411 progress.started_at = None;
412 progress.last_percent = None;
413 }
414 }
415}
416
417fn wait_for_link(
418 state: &Arc<(Mutex<ClientState>, Condvar)>,
419 link_id: [u8; 16],
420 timeout: Duration,
421) -> Result<()> {
422 let deadline = Instant::now() + timeout;
423 let (lock, cv) = &**state;
424 let mut state = lock.lock().unwrap();
425 loop {
426 if state.established.contains(&link_id) {
427 return Ok(());
428 }
429 let now = Instant::now();
430 if now >= deadline {
431 return Err(Error::msg("link establishment timed out"));
432 }
433 let (next, _) = cv
434 .wait_timeout(state, deadline.saturating_duration_since(now))
435 .unwrap();
436 state = next;
437 }
438}
439
440pub(crate) fn decode_status(bytes: Vec<u8>) -> Result<Vec<u8>> {
441 let Some((&code, body)) = bytes.split_first() else {
442 return Err(Error::msg("empty response"));
443 };
444 if code == protocol::RES_OK {
445 Ok(body.to_vec())
446 } else {
447 Err(Error::msg(format!(
448 "remote returned status 0x{code:02x}: {}",
449 String::from_utf8_lossy(body)
450 )))
451 }
452}
453
454pub(crate) fn ensure_metadata_ok(metadata: &[u8]) -> Result<()> {
455 let value = msgpack::unpack_exact(metadata)
456 .map_err(|e| Error::msg(format!("invalid response metadata: {e}")))?;
457 let Some(map) = value.as_map() else {
458 return Err(Error::msg("response metadata is not a map"));
459 };
460 let code = map.iter().find_map(|(key, value)| {
461 if matches!(key, Value::UInt(v) if *v == protocol::IDX_RESULT_CODE) {
462 value.as_uint().map(|v| v as u8)
463 } else {
464 None
465 }
466 });
467 match code {
468 Some(protocol::RES_OK) => Ok(()),
469 Some(code) => Err(Error::msg(format!("remote returned status 0x{code:02x}"))),
470 None => Err(Error::msg("response metadata missing status code")),
471 }
472}
473
474#[derive(Debug, Clone, PartialEq, Eq)]
475struct FetchRef {
476 sha: String,
477 name: String,
478}
479
480fn parse_fetch_command(input: &str) -> Result<FetchRef> {
481 let mut parts = input.split_whitespace();
482 let sha = parts
483 .next()
484 .ok_or_else(|| Error::msg("fetch command missing sha"))?;
485 let name = parts
486 .next()
487 .ok_or_else(|| Error::msg("fetch command missing ref"))?;
488 Ok(FetchRef {
489 sha: sha.to_string(),
490 name: name.to_string(),
491 })
492}
493
494#[derive(Debug, Clone, PartialEq, Eq)]
495struct PushSpec {
496 local: String,
497 remote: String,
498 force: bool,
499}
500
501fn parse_push_spec(input: &str) -> Result<PushSpec> {
502 let (force, spec) = input
503 .strip_prefix('+')
504 .map(|s| (true, s))
505 .unwrap_or((false, input));
506 let (local, remote) = spec
507 .split_once(':')
508 .ok_or_else(|| Error::msg("push spec must be <local>:<remote>"))?;
509 Ok(PushSpec {
510 local: local.to_string(),
511 remote: remote.to_string(),
512 force,
513 })
514}
515
516#[derive(Debug, Default)]
517struct ClientOptions {
518 config_dir: Option<PathBuf>,
519 rns_config_dir: Option<PathBuf>,
520 url: String,
521}
522
523impl ClientOptions {
524 fn parse<I>(args: I) -> Result<Self>
525 where
526 I: IntoIterator<Item = String>,
527 {
528 let mut options = ClientOptions::default();
529 let mut positional = Vec::new();
530 let mut args = args.into_iter();
531 while let Some(arg) = args.next() {
532 match arg.as_str() {
533 "-c" | "--config" => {
534 options.config_dir = Some(PathBuf::from(
535 args.next()
536 .ok_or_else(|| Error::msg("missing config path"))?,
537 ));
538 }
539 "--rnsconfig" => {
540 options.rns_config_dir = Some(PathBuf::from(
541 args.next()
542 .ok_or_else(|| Error::msg("missing RNS config path"))?,
543 ));
544 }
545 "-h" | "--help" => return Err(Error::msg(usage())),
546 other => positional.push(other.to_string()),
547 }
548 }
549 options.url = positional
550 .last()
551 .cloned()
552 .ok_or_else(|| Error::msg(usage()))?;
553 Ok(options)
554 }
555}
556
557fn usage() -> &'static str {
558 "usage: git-remote-rns [--config DIR] [--rnsconfig DIR] <remote-name> <rns://destination/repo>"
559}
560
561#[cfg(test)]
562mod tests {
563 use super::*;
564
565 #[test]
566 fn parses_fetch_command() {
567 assert_eq!(
568 parse_fetch_command("abc refs/heads/main").unwrap(),
569 FetchRef {
570 sha: "abc".into(),
571 name: "refs/heads/main".into()
572 }
573 );
574 }
575
576 #[test]
577 fn parses_forced_push_spec() {
578 assert_eq!(
579 parse_push_spec("+refs/heads/main:refs/heads/main").unwrap(),
580 PushSpec {
581 local: "refs/heads/main".into(),
582 remote: "refs/heads/main".into(),
583 force: true
584 }
585 );
586 }
587
588 #[test]
589 fn decodes_status_payload() {
590 assert_eq!(
591 decode_status(protocol::status_bytes(protocol::RES_OK, b"refs")).unwrap(),
592 b"refs"
593 );
594 assert!(decode_status(protocol::status_bytes(protocol::RES_NOT_FOUND, b"no")).is_err());
595 }
596
597 #[test]
598 fn metadata_status_ok_is_accepted() {
599 assert!(ensure_metadata_ok(&protocol::metadata_status(protocol::RES_OK)).is_ok());
600 assert!(ensure_metadata_ok(&protocol::metadata_status(protocol::RES_REMOTE_FAIL)).is_err());
601 }
602
603 #[test]
604 fn parses_remote_refs_and_ignores_head() {
605 let refs = parse_remote_refs(
606 b"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa HEAD\nbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb refs/heads/main\nmalformed\n",
607 );
608
609 assert_eq!(refs.len(), 1);
610 assert_eq!(
611 refs["refs/heads/main"],
612 "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
613 );
614 }
615
616 #[test]
617 fn fetch_have_set_uses_per_ref_and_global_matching_haves() {
618 let remote_refs = HashMap::from([
619 (
620 "refs/heads/main".to_string(),
621 "1111111111111111111111111111111111111111".to_string(),
622 ),
623 (
624 "refs/tags/v1".to_string(),
625 "2222222222222222222222222222222222222222".to_string(),
626 ),
627 ]);
628 let fetch_refs = vec![FetchRef {
629 sha: "3333333333333333333333333333333333333333".into(),
630 name: "refs/heads/feature".into(),
631 }];
632
633 let have = build_fetch_have_set(&remote_refs, &fetch_refs, |name| match name {
634 "refs/heads/main" => Some("1111111111111111111111111111111111111111".into()),
635 "refs/tags/v1" => Some("not-remote".into()),
636 "refs/heads/feature" => Some("4444444444444444444444444444444444444444".into()),
637 _ => None,
638 });
639
640 assert_eq!(
641 have,
642 vec![
643 "1111111111111111111111111111111111111111",
644 "4444444444444444444444444444444444444444"
645 ]
646 );
647 }
648
649 #[test]
650 fn push_exclusion_set_keeps_only_local_known_remote_objects() {
651 let remote_refs = HashMap::from([
652 ("refs/heads/main".to_string(), "aaaa".to_string()),
653 ("refs/heads/dev".to_string(), "bbbb".to_string()),
654 ("refs/tags/v1".to_string(), "aaaa".to_string()),
655 ]);
656
657 let exclusions = build_push_exclusion_set(&remote_refs, |sha| sha == "aaaa");
658
659 assert_eq!(exclusions, vec!["aaaa"]);
660 }
661}