netget/client/rsync/
mod.rs

1//! rsync client: each model action is one connection to the daemon (rsync opens a connection
2//! per transfer too) — the @RSYNCD handshake, the arguments, the file list, the requests and
3//! the end-of-run exchange, at protocol 29 (`src/server/rsync/wire.rs`).
4pub mod actions;
5
6use crate::client::llm_budget::call_llm_for_client;
7use crate::llm::actions::client_trait::{Client, ClientActionResult};
8use crate::logging::emit::Log;
9use crate::protocol::{ConnectContext, Event, EventType};
10use crate::server::rsync::wire::{self, Entry, Kind, MuxReader, Request};
11use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
12use crate::state::{AccessLogOwner, ClientStatus};
13pub use actions::RsyncClientProtocol;
14use anyhow::{bail, ensure, Context, Result};
15use serde_json::{json, Value};
16use std::net::SocketAddr;
17use std::time::Duration;
18use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
19use tokio::net::TcpStream;
20use tokio::sync::{mpsc, oneshot};
21
22/// A whole operation, connect to close.
23pub const OPERATION_TIMEOUT: Duration = Duration::from_secs(120);
24/// How many model turns may follow from one another before the chain stops.
25pub const MAX_FOLLOWUP_DEPTH: u32 = 8;
26const TURNS: usize = 64;
27
28#[derive(Clone)]
29struct Settings {
30    daemon: SocketAddr,
31    auth: Option<(String, String)>,
32    max_fetch: u64,
33}
34
35pub async fn connect(ctx: ConnectContext) -> Result<SocketAddr> {
36    let params = ctx.startup_params.as_ref();
37    let get = |k: &str| -> Result<Option<String>> {
38        Ok(params
39            .map(|p| p.get_optional_string(k))
40            .transpose()?
41            .flatten())
42    };
43    let auth = match (get("username")?, get("password")?) {
44        (Some(u), Some(p)) => {
45            ensure!(!u.contains([' ', '\n', '\r']), "username is one word");
46            Some((u, p))
47        }
48        (None, None) => None,
49        _ => bail!("username and password are given together"),
50    };
51    let max_fetch = params
52        .map(|p| p.get_optional_u64("max_fetch_bytes"))
53        .transpose()?
54        .flatten()
55        .unwrap_or(actions::DEFAULT_MAX_FETCH_BYTES);
56    ensure!(
57        (1..=actions::MAX_FETCH_BYTES_LIMIT).contains(&max_fetch),
58        "max_fetch_bytes must be between 1 and {}",
59        actions::MAX_FETCH_BYTES_LIMIT
60    );
61    let raw = ctx
62        .remote_addr
63        .trim()
64        .trim_start_matches("rsync://")
65        .trim_end_matches('/');
66    let daemon = match raw.parse::<SocketAddr>() {
67        Ok(a) => a,
68        Err(_) => {
69            let with_port = if raw.contains(':') {
70                raw.to_string()
71            } else {
72                format!("{raw}:873")
73            };
74            let found = tokio::net::lookup_host(&with_port)
75                .await
76                .with_context(|| format!("cannot resolve {with_port}"))?
77                .next()
78                .with_context(|| format!("{with_port} resolved to no address"))?;
79            found
80        }
81    };
82    let settings = Settings {
83        daemon,
84        auth,
85        max_fetch,
86    };
87    ctx.state
88        .update_client_status(ctx.client_id, ClientStatus::Connected)
89        .await;
90    Log::new(Some(&ctx.status_tx)).info(format!(
91        "rsync client {} ready for rsync://{daemon}/",
92        ctx.client_id
93    ));
94    let external =
95        crate::client::command_support::register_command_channel(&ctx.state, ctx.client_id).await;
96    let (event_tx, event_rx) = mpsc::channel::<(Event, u32)>(TURNS);
97    let (internal_tx, internal) = mpsc::channel::<(Value, u32)>(64);
98    let _ = event_tx.try_send((
99        Event::new(&actions::READY_EVENT, json!({"daemon": daemon.to_string()})),
100        0,
101    ));
102    let dispatcher = tokio::spawn(run_turns(ctx.clone(), event_rx, internal_tx));
103    let dispatcher_abort = dispatcher.abort_handle();
104    ctx.state
105        .register_client_task(ctx.client_id, dispatcher)
106        .await;
107    let session_ctx = ctx.clone();
108    let task = tokio::spawn(async move {
109        run(&session_ctx, settings, external, internal, event_tx).await;
110        dispatcher_abort.abort();
111        session_ctx
112            .state
113            .update_client_status(session_ctx.client_id, ClientStatus::Disconnected)
114            .await;
115        session_ctx
116            .state
117            .remove_client_handle(session_ctx.client_id)
118            .await;
119        let _ = session_ctx.status_tx.send("__UPDATE_UI__".into());
120    });
121    ctx.state.register_client_task(ctx.client_id, task).await;
122    Ok(daemon)
123}
124
125async fn run(
126    ctx: &ConnectContext,
127    settings: Settings,
128    mut external: mpsc::Receiver<ClientCommand>,
129    mut internal: mpsc::Receiver<(Value, u32)>,
130    events: mpsc::Sender<(Event, u32)>,
131) {
132    loop {
133        let (action, depth, command) = tokio::select! {
134            c = external.recv() => match c {
135                Some(c) => {
136                    ctx.state.record_access_log(AccessLogOwner::Client(ctx.client_id.as_u32()), "rsync", None,
137                        "injected_action", c.action.clone(), vec![]).await;
138                    (c.action.clone(), 0, Some(c))
139                }
140                None => return,
141            },
142            a = internal.recv() => match a {
143                Some((a, d)) => (a, d, None),
144                None => return,
145            },
146        };
147        let reply = |command: Option<ClientCommand>, outcome: ClientSendOutcome| {
148            if let Some(c) = command {
149                crate::client::command_support::reply(c, Ok(outcome));
150            }
151        };
152        let target = match RsyncClientProtocol.execute_action(action.clone()) {
153            Ok(ClientActionResult::Disconnect) => {
154                reply(command, ClientSendOutcome::Disconnected);
155                return;
156            }
157            Ok(_) => actions::check(&action).ok().flatten(),
158            Err(e) => {
159                reply(
160                    command,
161                    ClientSendOutcome::Rejected {
162                        error: e.to_string(),
163                    },
164                );
165                continue;
166            }
167        };
168        let kind = action["type"].as_str().unwrap_or_default().to_string();
169        let recursive = action["recursive"].as_bool().unwrap_or(false);
170        let result = tokio::time::timeout(OPERATION_TIMEOUT, async {
171            if kind == "rsync_push" {
172                push(&settings, target.clone(), &action).await
173            } else {
174                operation(&settings, &kind, target.clone(), recursive).await
175            }
176        })
177        .await;
178        let (event, data): (&'static EventType, Value) = match result {
179            Ok(Ok(v)) => v,
180            Ok(Err(e)) => (
181                &actions::ERROR_EVENT,
182                json!({"operation": kind, "path": target.map(|t| t.1), "message": format!("{e:#}")}),
183            ),
184            Err(_) => (
185                &actions::ERROR_EVENT,
186                json!({"operation": kind, "path": target.map(|t| t.1), "message": "the operation timed out"}),
187            ),
188        };
189        Log::new(Some(&ctx.status_tx)).info(format!(
190            "rsync client {}: {kind} -> {}",
191            ctx.client_id, event.id
192        ));
193        reply(
194            command,
195            ClientSendOutcome::Executed {
196                detail: data.to_string(),
197            },
198        );
199        if events.try_send((Event::new(event, data), depth)).is_err() {
200            tracing::warn!("rsync client {} dropped an event: the model is {TURNS} behind decision=turn_queue_full", ctx.client_id);
201        }
202    }
203}
204
205/// The text phase: greetings, the module line, MOTD lines, and authentication.
206/// Returns the lines that were neither status nor error (listing lines, MOTD).
207async fn handshake(
208    rd: &mut BufReader<tokio::net::tcp::OwnedReadHalf>,
209    wr: &mut tokio::net::tcp::OwnedWriteHalf,
210    module_line: &str,
211    auth: &Option<(String, String)>,
212) -> Result<(bool, Vec<String>)> {
213    wr.write_all(wire::GREETING.as_bytes()).await?;
214    let mut lines = Vec::new();
215    let greeting = read_line(rd).await?;
216    ensure!(
217        greeting.starts_with("@RSYNCD: "),
218        "not an rsync daemon: {greeting:?}"
219    );
220    wr.write_all(format!("{module_line}\n").as_bytes()).await?;
221    loop {
222        let line = read_line(rd).await?;
223        if line == "@RSYNCD: OK" {
224            return Ok((true, lines));
225        }
226        if line == "@RSYNCD: EXIT" {
227            return Ok((false, lines));
228        }
229        if let Some(e) = line.strip_prefix("@ERROR") {
230            bail!("{}", e.trim_start_matches(':').trim());
231        }
232        if let Some(challenge) = line.strip_prefix("@RSYNCD: AUTHREQD ") {
233            let Some((user, password)) = auth else {
234                bail!("the module requires authentication and no username/password is configured");
235            };
236            wr.write_all(
237                format!("{user} {}\n", wire::auth_response(password, challenge)).as_bytes(),
238            )
239            .await?;
240            continue;
241        }
242        lines.push(line);
243    }
244}
245
246async fn read_line(rd: &mut BufReader<tokio::net::tcp::OwnedReadHalf>) -> Result<String> {
247    let mut buf = Vec::new();
248    let n = tokio::io::AsyncReadExt::take(&mut *rd, wire::MAX_LINE as u64)
249        .read_until(b'\n', &mut buf)
250        .await?;
251    ensure!(n > 0, "the daemon closed the connection");
252    ensure!(
253        buf.ends_with(b"\n"),
254        "a line longer than {} bytes",
255        wire::MAX_LINE
256    );
257    Ok(String::from_utf8_lossy(&buf)
258        .trim_end_matches(['\n', '\r'])
259        .to_string())
260}
261
262async fn operation(
263    s: &Settings,
264    kind: &str,
265    target: Option<(String, String)>,
266    recursive: bool,
267) -> Result<(&'static EventType, Value)> {
268    let stream = tokio::time::timeout(Duration::from_secs(10), TcpStream::connect(s.daemon))
269        .await
270        .context("connecting timed out")??;
271    let (rd, mut wr) = stream.into_split();
272    let mut rd = BufReader::new(rd);
273    let Some((module, arg)) = target else {
274        let (_, lines) = handshake(&mut rd, &mut wr, "", &None).await?;
275        let mut modules = Vec::new();
276        let mut motd = Vec::new();
277        for l in lines {
278            match l.split_once('\t') {
279                Some((name, comment)) => {
280                    modules.push(json!({"name": name.trim(), "comment": comment.trim()}))
281                }
282                None => motd.push(l),
283            }
284        }
285        return Ok((
286            &actions::MODULES_EVENT,
287            json!({"modules": modules, "motd": motd.join("\n")}),
288        ));
289    };
290    let (ok, _motd) = handshake(&mut rd, &mut wr, &module, &s.auth).await?;
291    ensure!(
292        ok,
293        "the daemon ended the session before accepting the module"
294    );
295    let req = Request {
296        sender: true,
297        recursive,
298        dirs: !recursive,
299        links: true,
300        ..Default::default()
301    };
302    let flags = if recursive { "-lr" } else { "-ld" };
303    let args = format!("--server\n--sender\n{flags}\n.\n{arg}\n\n");
304    wr.write_all(args.as_bytes()).await?;
305    let mut seed = [0u8; 4];
306    tokio::io::AsyncReadExt::read_exact(&mut rd, &mut seed)
307        .await
308        .context("no checksum seed")?;
309    let seed = i32::from_le_bytes(seed);
310    wr.write_all(&wire::le32(0)).await?; // no filter rules
311    let mut mux = MuxReader::new(rd);
312    let (mut entries, io_error) = mux.file_list(&req).await?;
313    wire::sort(&mut entries);
314    let errors: Vec<String> = mux
315        .messages
316        .iter()
317        .filter(|(t, _)| *t == wire::MSG_ERROR_XFER || *t == wire::MSG_ERROR)
318        .map(|(_, m)| m.clone())
319        .collect();
320    if entries.is_empty() {
321        // The daemon closes after an empty list.
322        let message = if errors.is_empty() {
323            "nothing at that path".to_string()
324        } else {
325            errors.join("; ")
326        };
327        bail!("{message}");
328    }
329
330    // Which files to ask for: regular files, while the budget lasts.
331    let fetch = kind == "rsync_fetch";
332    let mut wanted = Vec::new();
333    let mut skipped = Vec::new();
334    let mut budget = s.max_fetch;
335    if fetch {
336        for (i, e) in entries.iter().enumerate() {
337            if e.kind != Kind::File {
338                continue;
339            }
340            if e.size <= budget {
341                budget -= e.size;
342                wanted.push(i);
343            } else {
344                skipped.push(e.path.clone());
345            }
346        }
347    }
348    let (first_echo_tx, first_echo) = oneshot::channel::<()>();
349    let (stats_tx, stats_read) = oneshot::channel::<()>();
350    let writer = async {
351        let mut requests = Vec::with_capacity(wanted.len() * 22 + 4);
352        for &i in &wanted {
353            requests.extend_from_slice(&wire::le32(i as i32));
354            requests.extend_from_slice(&(wire::ITEM_TRANSFER | wire::ITEM_IS_NEW).to_le_bytes());
355            requests.extend_from_slice(&[0u8; 16]); // no basis: send everything
356        }
357        requests.extend_from_slice(&wire::le32(wire::NDX_DONE));
358        wr.write_all(&requests).await?;
359        first_echo.await.context("the reply stream ended early")?;
360        wr.write_all(&[wire::le32(wire::NDX_DONE), wire::le32(wire::NDX_DONE)].concat())
361            .await?;
362        stats_read.await.context("the reply stream ended early")?;
363        wr.write_all(&wire::le32(wire::NDX_DONE)).await?; // the final goodbye
364        let _ = wr.shutdown().await;
365        anyhow::Ok(())
366    };
367    let reader = async {
368        let mut files = Vec::new();
369        let mut first_echo_tx = Some(first_echo_tx);
370        let mut phase = 0;
371        loop {
372            let ndx = mux.int().await?;
373            if ndx == wire::NDX_DONE {
374                phase += 1;
375                if let Some(tx) = first_echo_tx.take() {
376                    let _ = tx.send(());
377                }
378                if phase == 3 {
379                    break;
380                }
381                continue;
382            }
383            let iflags = mux.short().await?;
384            if iflags & wire::ITEM_BASIS_TYPE_FOLLOWS != 0 {
385                mux.byte().await?;
386            }
387            if iflags & wire::ITEM_XNAME_FOLLOWS != 0 {
388                let mut n = mux.byte().await? as usize;
389                if n & 0x80 != 0 {
390                    n = ((n & 0x7f) << 8) | mux.byte().await? as usize;
391                }
392                mux.read(n).await?;
393            }
394            if iflags & wire::ITEM_TRANSFER == 0 {
395                continue;
396            }
397            ensure!(
398                ndx >= 0 && (ndx as usize) < entries.len(),
399                "a reply for index {ndx}"
400            );
401            for _ in 0..4 {
402                mux.int().await?;
403            }
404            let e: &Entry = &entries[ndx as usize];
405            let mut data = Vec::new();
406            loop {
407                let t = mux.int().await?;
408                if t == 0 {
409                    break;
410                }
411                ensure!(t > 0, "a block-match token with no basis file");
412                ensure!(
413                    data.len() as u64 + t as u64 <= e.size.max(1) * 2 + wire::CHUNK as u64,
414                    "more data than the file list said"
415                );
416                data.extend(mux.read(t as usize).await?);
417            }
418            let sum = mux.read(16).await?;
419            ensure!(
420                sum == wire::file_sum(seed, &data),
421                "{}: checksum mismatch",
422                e.path
423            );
424            let (content, encoding) = wire::content_json(&data);
425            files.push(json!({"path": e.path, "size": data.len(), "content": content, "encoding": encoding}));
426        }
427        for _ in 0..5 {
428            mux.longint().await?;
429        }
430        let _ = stats_tx.send(());
431        anyhow::Ok(files)
432    };
433    let (w, files) = tokio::join!(writer, reader);
434    let files = files?;
435    w?;
436    let errors: Vec<String> = mux
437        .messages
438        .iter()
439        .filter(|(t, _)| *t == wire::MSG_ERROR_XFER || *t == wire::MSG_ERROR)
440        .map(|(_, m)| m.clone())
441        .collect();
442    ensure!(
443        errors.is_empty() && io_error == 0 && mux.io_error == 0,
444        "{}",
445        if errors.is_empty() {
446            "the daemon reported an I/O error".into()
447        } else {
448            errors.join("; ")
449        }
450    );
451    let listing: Vec<Value> = entries.iter().map(wire::entry_json).collect();
452    if fetch {
453        Ok((
454            &actions::FETCHED_EVENT,
455            json!({"path": arg, "files": files, "entries": listing, "skipped": skipped}),
456        ))
457    } else {
458        Ok((
459            &actions::LISTING_EVENT,
460            json!({"path": arg, "entries": listing}),
461        ))
462    }
463}
464
465/// A push: this client is rsync's sender, the daemon its receiver (protocol 29).
466///
467/// The file list goes out raw, built from the action's files plus the directories they imply
468/// and "." for the destination itself. The daemon's generator then asks for files by index —
469/// multiplexed, each with an item-flags word and a checksum header — and every one is answered
470/// whole: the same header echoed, the data as literal tokens, MD4(seed ‖ data). Items it only
471/// reports (a directory it made) are echoed back as rsync's sender does. Each NDX_DONE is
472/// echoed; after the third comes the daemon's final one, and the connection closes.
473async fn push(
474    s: &Settings,
475    target: Option<(String, String)>,
476    action: &Value,
477) -> Result<(&'static EventType, Value)> {
478    let (module, arg) = target.context("a push needs a destination")?;
479    let now = crate::utils::clock::SystemTime::now()
480        .duration_since(crate::utils::clock::UNIX_EPOCH)
481        .map(|d| d.as_secs() as u32)
482        .unwrap_or(0);
483    let mut entries = Vec::new();
484    for f in action["files"].as_array().into_iter().flatten() {
485        entries.push(wire::entry_from_json(f, now)?);
486    }
487    crate::server::rsync::complete_dirs(&mut entries, now);
488    let mut root = wire::entry_from_json(&json!({"path": "root", "type": "dir"}), now)?;
489    root.path = ".".into();
490    entries.push(root);
491    wire::sort(&mut entries);
492    let req = Request {
493        recursive: true,
494        links: true,
495        ..Default::default()
496    };
497    let mut list = Vec::new();
498    let mut writer = wire::FlistWriter::default();
499    for e in &entries {
500        writer.entry(e, e.path == ".", &req, &mut list);
501    }
502    wire::FlistWriter::finish(&req, 0, &mut list);
503
504    let stream = tokio::time::timeout(Duration::from_secs(10), TcpStream::connect(s.daemon))
505        .await
506        .context("connecting timed out")??;
507    let (rd, mut wr) = stream.into_split();
508    let mut rd = BufReader::new(rd);
509    let (ok, _motd) = handshake(&mut rd, &mut wr, &module, &s.auth).await?;
510    ensure!(
511        ok,
512        "the daemon ended the session before accepting the module"
513    );
514    wr.write_all(format!("--server\n-lr\n.\n{arg}\n\n").as_bytes())
515        .await?;
516    let mut seed = [0u8; 4];
517    tokio::io::AsyncReadExt::read_exact(&mut rd, &mut seed)
518        .await
519        .context("no checksum seed")?;
520    let seed = i32::from_le_bytes(seed);
521    wr.write_all(&list).await?;
522
523    let mut mux = MuxReader::new(rd);
524    let mut sent = Vec::new();
525    let mut requested = std::collections::HashSet::new();
526    let mut dones = 0;
527    let result: Result<()> = async {
528        loop {
529            let ndx = mux.int().await?;
530            if ndx == wire::NDX_DONE {
531                wr.write_all(&wire::le32(wire::NDX_DONE)).await?;
532                dones += 1;
533                if dones == 3 {
534                    break;
535                }
536                continue;
537            }
538            if ndx as usize == entries.len() {
539                // A protocol-29 keep-alive from a generator busy on something slow.
540                mux.short().await?;
541                continue;
542            }
543            ensure!(
544                ndx >= 0 && (ndx as usize) < entries.len(),
545                "the daemon asked for index {ndx} of {}",
546                entries.len()
547            );
548            let iflags = mux.short().await?;
549            let mut reply = wire::le32(ndx).to_vec();
550            reply.extend_from_slice(&iflags.to_le_bytes());
551            if iflags & wire::ITEM_BASIS_TYPE_FOLLOWS != 0 {
552                reply.push(mux.byte().await?);
553            }
554            if iflags & wire::ITEM_XNAME_FOLLOWS != 0 {
555                let first = mux.byte().await?;
556                reply.push(first);
557                let mut n = first as usize;
558                if n & 0x80 != 0 {
559                    let low = mux.byte().await?;
560                    reply.push(low);
561                    n = ((n & 0x7f) << 8) | low as usize;
562                }
563                reply.extend(mux.read(n).await?);
564            }
565            if iflags & wire::ITEM_TRANSFER == 0 {
566                wr.write_all(&reply).await?;
567                continue;
568            }
569            let entry = &entries[ndx as usize];
570            ensure!(
571                entry.kind == Kind::File,
572                "the daemon asked to transfer {:?}, which is not a file",
573                entry.path
574            );
575            let mut head = [0i32; 4];
576            for v in head.iter_mut() {
577                *v = mux.int().await?;
578            }
579            let [count, _blength, s2length, _remainder] = head;
580            ensure!(
581                (0..=wire::MAX_SUM_COUNT).contains(&count) && (0..=16).contains(&s2length),
582                "a malformed checksum header"
583            );
584            // Block checksums of the daemon's existing copy: read, and not used, since every
585            // file is sent whole.
586            mux.read(count as usize * (4 + s2length as usize)).await?;
587            for v in head {
588                reply.extend_from_slice(&wire::le32(v));
589            }
590            for chunk in entry.data.chunks(wire::CHUNK) {
591                reply.extend_from_slice(&wire::le32(chunk.len() as i32));
592                reply.extend_from_slice(chunk);
593            }
594            reply.extend_from_slice(&wire::le32(0));
595            reply.extend_from_slice(&wire::file_sum(seed, &entry.data));
596            wr.write_all(&reply).await?;
597            if requested.insert(ndx) {
598                sent.push(json!({"path": entry.path, "size": entry.data.len()}));
599            }
600        }
601        // The daemon's final NDX_DONE, then it closes.
602        ensure!(
603            mux.int().await? == wire::NDX_DONE,
604            "no final goodbye from the daemon"
605        );
606        Ok(())
607    }
608    .await;
609    let _ = wr.shutdown().await;
610    let errors: Vec<String> = mux
611        .messages
612        .iter()
613        .filter(|(t, _)| *t == wire::MSG_ERROR_XFER || *t == wire::MSG_ERROR)
614        .map(|(_, m)| m.clone())
615        .collect();
616    if let Err(e) = result {
617        // A daemon that refuses outright says why and closes; that is the answer.
618        if errors.is_empty() {
619            return Err(e);
620        }
621        bail!("{}", errors.join("; "));
622    }
623    let not_requested: Vec<String> = entries
624        .iter()
625        .enumerate()
626        .filter(|(i, e)| e.kind == Kind::File && !requested.contains(&(*i as i32)))
627        .map(|(_, e)| e.path.clone())
628        .collect();
629    let complete = errors.is_empty() && mux.io_error == 0;
630    Ok((
631        &actions::PUSHED_EVENT,
632        json!({"path": arg, "sent": sent, "not_requested": not_requested, "errors": errors, "complete": complete}),
633    ))
634}
635
636async fn run_turns(
637    ctx: ConnectContext,
638    mut events: mpsc::Receiver<(Event, u32)>,
639    internal: mpsc::Sender<(Value, u32)>,
640) {
641    let protocol = RsyncClientProtocol;
642    while let Some((event, depth)) = events.recv().await {
643        if depth >= MAX_FOLLOWUP_DEPTH {
644            tracing::warn!("rsync client {} not asking the model about {}: {depth} turns deep decision=followup_depth", ctx.client_id, event.id());
645            continue;
646        }
647        let instruction = ctx
648            .state
649            .get_instruction_for_client(ctx.client_id)
650            .await
651            .unwrap_or_default();
652        let memory = ctx
653            .state
654            .get_memory_for_client(ctx.client_id)
655            .await
656            .unwrap_or_default();
657        match call_llm_for_client(
658            &ctx.llm_client,
659            &ctx.state,
660            ctx.client_id.to_string(),
661            &instruction,
662            &memory,
663            Some(&event),
664            &protocol,
665            &ctx.status_tx,
666        )
667        .await
668        {
669            Ok(result) => {
670                if let Some(memory) = result.memory_updates {
671                    ctx.state.set_memory_for_client(ctx.client_id, memory).await;
672                }
673                for action in result.actions {
674                    if internal.send((action, depth + 1)).await.is_err() {
675                        return;
676                    }
677                }
678            }
679            Err(e) => Log::new(Some(&ctx.status_tx)).warn(format!("rsync client handler: {e}")),
680        }
681    }
682}
netget/client/rsync/mod.rs

netget/client/rsync/
mod.rs
