wsb-daemon 0.248.6265

//! Serving what a peer asks, and the rules for answering.
//!
//! Every request is answered from what THIS machine can verify. A peer says
//! what it wants; it never says what this side must believe. In particular:
//!
//! * an `Apply` is checked against this side's own current git epoch before a
//!   byte is written — the sender's epoch is a claim, and the claim is only
//!   useful because it must MATCH what is found here;
//! * every path is resolved through the one function that refuses `.git`,
//!   `.wsb` and anything outside the repository;
//! * a workspace or repository this side does not have is a refusal with a
//!   code, not an error — the other machine simply has something we do not.

use std::sync::Arc;

use wsb::peer::compatibility::RepositoryFacts;
use wsb::peer::converge::{Baseline, EntryKind};
use wsb::peer::model::PeerId;
use wsb::peer::protocol::{
    baseline_digest, RefusalCode, Request, Response, Write, WorkspaceSummary,
};
use wsb::peer::scan;
use wsb::peer::transport::Session;

use super::coordinator::LocalWorkspace;
use super::service::now;
use super::store::MachineStore;

/// What a served connection needs to reach.
pub trait Machine: Send + Sync {
    /// Every workspace this machine holds, for the inventory.
    fn workspaces(&self) -> Vec<WorkspaceSummary>;
    /// The repository tree of the worktree a peer means by `id`, or why this
    /// machine cannot answer for one: it has none, or two claim it.
    fn workspace(&self, id: &wsb::peer::model::WorkspaceId) -> Result<Arc<LocalWorkspace>, String>;
    fn store(&self) -> Arc<MachineStore>;
    fn staging_root(&self) -> std::path::PathBuf;
    /// The plugins running here, for a peer's request to one of them.
    fn plugins(&self) -> Arc<crate::plugin::PluginRegistry>;
}

/// Serve one peer's requests until it goes away.
///
/// Returns when the peer closes the connection, which is ordinary. Any other
/// end is an error the caller reports against that peer.
pub fn serve(machine: &dyn Machine, session: &mut Session, peer: &PeerId) -> Result<(), String> {
    loop {
        let request: Request = match session.receive() {
            Ok(request) => request,
            // A closed connection reads as an error at this layer; it is the
            // ordinary end of a conversation, not a failure to report.
            Err(error) if is_disconnect(&error) => return Ok(()),
            Err(error) => return Err(error),
        };
        let response = answer(machine, peer, request);
        session.send(&response)?;
    }
}

fn is_disconnect(error: &str) -> bool {
    error.contains("cannot read from the peer")
}

/// Answer one request.
pub fn answer(machine: &dyn Machine, peer: &PeerId, request: Request) -> Response {
    match request {
        Request::Inventory => Response::Inventory {
            workspaces: machine.workspaces(),
        },

        Request::Report {
            workspace,
            repositories,
        } => match machine.workspace(&workspace) {
            Err(_) => Response::Report {
                workspace_present: Some(false),
                repositories: Vec::new(),
            },
            Ok(local) => Response::Report {
                workspace_present: Some(true),
                repositories: local.report(&repositories),
            },
        },

        Request::Fetch {
            workspace,
            repository,
            epoch,
            paths,
        } => {
            // Continuity must be enabled HERE to give files away, as it must
            // be to take them: a machine that paused carries nothing either
            // way. A peer whose pass began before the pause would otherwise
            // still fetch what was written after it.
            match machine.store().desired(peer, &workspace) {
                Ok(wsb::peer::model::Desired::Enabled) => {}
                Ok(_) => {
                    return refuse(
                        RefusalCode::NotEnabled,
                        "continuity is not enabled for this workspace on this machine",
                    )
                }
                Err(error) => return refuse(RefusalCode::Internal, &error),
            }
            let local = match machine.workspace(&workspace) {
                Ok(local) => local,
                Err(why) => return refuse(RefusalCode::UnknownWorkspace, &why),
            };
            let Some(repo) = local.repository(&repository) else {
                return refuse(RefusalCode::UnknownRepository, "this machine does not have that repository");
            };
            // The epoch is checked BEFORE the files are read: content read
            // under one git state and sent after another is exactly the mix
            // this feature must never produce.
            match RepositoryFacts::read(repository.clone(), &repo.workdir, &repo.git_dir) {
                Ok(facts) if epoch.still_current(&facts) => {}
                Ok(_) => return refuse(RefusalCode::EpochMoved, "the repository moved"),
                Err(error) => return refuse(RefusalCode::Internal, &error),
            }

            let mut writes = Vec::new();
            for path in paths {
                let absolute = match scan::local_path(&repo.workdir, &path) {
                    Ok(absolute) => absolute,
                    Err(error) => return refuse(RefusalCode::ForbiddenPath, &error),
                };
                let entry = match wsb::peer::converge::Entry::read(&absolute) {
                    Ok(Some(entry)) => entry,
                    // Gone since the scan. Ordinary: the next round sees the
                    // deletion. Nothing is sent for it.
                    Ok(None) => continue,
                    Err(error) => return refuse(RefusalCode::Internal, &error),
                };
                let content = match entry.kind {
                    EntryKind::Symlink => match std::fs::read_link(&absolute) {
                        Ok(target) => {
                            wsb::peer::identity::base64_encode(scan::to_slash(&target).as_bytes())
                        }
                        Err(error) => return refuse(RefusalCode::Internal, &error.to_string()),
                    },
                    EntryKind::File => match std::fs::read(&absolute) {
                        Ok(bytes) => wsb::peer::identity::base64_encode(&bytes),
                        Err(error) => return refuse(RefusalCode::Internal, &error.to_string()),
                    },
                };
                writes.push(Write {
                    path,
                    entry,
                    content,
                });
            }
            Response::Fetched { writes }
        }

        Request::Confirm {
            workspace,
            repository,
            baseline_digest: asker_digest,
        } => {
            // Answered from this machine's own copy of the baseline, which is
            // the whole point: the asker has one too, and the question is
            // whether they are the same. A mismatch is refused with the code
            // that makes the asker reseed rather than act on its copy alone.
            let local = match machine.workspace(&workspace) {
                Ok(local) => local,
                Err(why) => return refuse(RefusalCode::UnknownWorkspace, &why),
            };
            let baseline = match machine.store().baseline(peer, &workspace, &repository) {
                Ok(baseline) => baseline,
                Err(error) => return refuse(RefusalCode::Internal, &error),
            };
            if baseline_digest(&baseline) != asker_digest {
                // BOTH sides re-seed, and that is what makes drift heal.
                //
                // A baseline is held twice, once per machine, and a mismatch is
                // evidence about the pair rather than about either one — this
                // side has exactly as much reason to distrust its copy as the
                // asker has to distrust theirs. Re-seeding only the asker left
                // the two at different states again: it seeds from the commit
                // they share, this side keeps whatever it had, and the next
                // pass mismatches for the same reason. The asker re-seeded in a
                // loop and nothing ever crossed.
                //
                // Both seeding from the shared commit is deterministic: the
                // same commit gives the same baseline on both machines, so one
                // round of this ends the disagreement.
                //
                // Re-seeding cannot lose a file. It drops carried-but-untracked
                // paths from the BASELINE, and a path in neither baseline that
                // both machines hold is agreement, not a deletion.
                // Re-seeding is this side's recovery, and failing at it must
                // not stop this side SAYING what it already knows. Answering
                // `Internal` here would leave the asker with an error instead
                // of the one code that makes it re-seed too, and the two would
                // go on disagreeing — refusing every transfer between them —
                // over a directory that could not be read.
                if let Err(error) = reseed_from_head(machine, peer, &local, &repository) {
                    tracing::warn!(
                        %peer,
                        repository = %repository,
                        %error,
                        "could not re-seed after a baseline mismatch; the peer is \
                         still told the baselines differ"
                    );
                }
                return refuse(
                    RefusalCode::BaselineMismatch,
                    "this machine holds a different baseline for that repository",
                );
            }
            Response::Acknowledged
        }

        Request::Apply {
            workspace,
            repository,
            epoch,
            writes,
            deletions,
            found,
            baseline_digest: sender_digest,
        } => {
            let store = machine.store();
            // Continuity must be enabled HERE too. A peer cannot start writing
            // into a workspace this machine never agreed to synchronise.
            match store.desired(peer, &workspace) {
                Ok(wsb::peer::model::Desired::Enabled) => {}
                Ok(_) => {
                    return refuse(
                        RefusalCode::NotEnabled,
                        "continuity is not enabled for this workspace on this machine",
                    )
                }
                Err(error) => return refuse(RefusalCode::Internal, &error),
            }

            let local = match machine.workspace(&workspace) {
                Ok(local) => local,
                Err(why) => return refuse(RefusalCode::UnknownWorkspace, &why),
            };
            let Some(repo) = local.repository(&repository) else {
                return refuse(RefusalCode::UnknownRepository, "this machine does not have that repository");
            };

            // The sender planned against a baseline. If this side holds a
            // different one, the plan was made from a state that has moved and
            // applying it would resolve a difference nobody computed.
            let baseline = match store.baseline(peer, &workspace, &repository) {
                Ok(baseline) => baseline,
                Err(error) => return refuse(RefusalCode::Internal, &error),
            };
            if baseline_digest(&baseline) != sender_digest {
                return refuse(
                    RefusalCode::BaselineMismatch,
                    "this machine holds a different baseline for that repository",
                );
            }

            match super::transfer::apply(
                &repo.workdir,
                &repository,
                &repo.git_dir,
                &epoch,
                &writes,
                &deletions,
                &found,
                &machine.staging_root(),
            ) {
                Ok(applied) => Response::Applied { applied },
                Err(super::transfer::ApplyRefusal::EpochMoved) => {
                    refuse(RefusalCode::EpochMoved, "the repository moved")
                }
                Err(super::transfer::ApplyRefusal::Forbidden(detail)) => {
                    refuse(RefusalCode::ForbiddenPath, &detail)
                }
                Err(super::transfer::ApplyRefusal::Malformed(detail)) => {
                    refuse(RefusalCode::Malformed, &detail)
                }
                Err(super::transfer::ApplyRefusal::Internal(detail)) => {
                    refuse(RefusalCode::Internal, &detail)
                }
            }
        }


        Request::Acknowledge {
            workspace,
            repository,
            epoch: _,
            paths,
            baseline_digest: sender_digest,
        } => {
            let store = machine.store();
            let local = match machine.workspace(&workspace) {
                Ok(local) => local,
                Err(why) => return refuse(RefusalCode::UnknownWorkspace, &why),
            };
            let Some(repo) = local.repository(&repository) else {
                return refuse(RefusalCode::UnknownRepository, "this machine does not have that repository");
            };
            let baseline = match store.baseline(peer, &workspace, &repository) {
                Ok(baseline) => baseline,
                Err(error) => return refuse(RefusalCode::Internal, &error),
            };
            if baseline_digest(&baseline) != sender_digest {
                return refuse(
                    RefusalCode::BaselineMismatch,
                    "this machine holds a different baseline for that repository",
                );
            }

            // The acknowledged paths are re-read from THIS side's disk rather
            // than taken from the message. What both machines agree on is what
            // is actually here now — an entry taken on trust would put a hash
            // in the baseline that this machine's own files do not have.
            let mut next: Baseline = baseline;
            for path in paths {
                let absolute = match scan::local_path(&repo.workdir, &path) {
                    Ok(absolute) => absolute,
                    Err(error) => return refuse(RefusalCode::ForbiddenPath, &error),
                };
                match wsb::peer::converge::Entry::read(&absolute) {
                    Ok(Some(entry)) => {
                        next.insert(path, entry);
                    }
                    Ok(None) => {
                        next.remove(&path);
                    }
                    Err(error) => return refuse(RefusalCode::Internal, &error),
                }
            }
            match store.put_baseline(peer, &workspace, &repository, &next) {
                Ok(()) => {
                    let _ = store.note_seen(peer, now());
                    Response::Acknowledged
                }
                Err(error) => refuse(RefusalCode::Internal, &error),
            }
        }

        Request::Plugin { service, body } => answer_plugin(machine, peer, service, body),
    }
}

/// Hand a peer's request to the plugin here that offers the service.
///
/// In this order, each a refusal the asking plugin can act on: the body must
/// be within bounds; the service must be switched on with this peer on THIS
/// side, which is the owner's decision and nothing the peer can change; and a
/// running plugin must offer it. What the plugin is told about the caller is
/// what this machine proved during the handshake and recorded at pairing,
/// never anything the request says about itself. The body is passed on
/// unread and is never logged.
fn answer_plugin(
    machine: &dyn Machine,
    peer: &PeerId,
    service: wsb::plugin::PluginService,
    body: String,
) -> Response {
    if body.len() > crate::plugin::encoded_bound() {
        return refuse(
            RefusalCode::Malformed,
            &format!("a plugin request may carry at most {} bytes", wsb::plugin::MAX_BODY),
        );
    }
    let store = machine.store();
    let switch = match store.service(peer, wsb::peer::model::Service::Plugin(service.clone())) {
        Ok(switch) => switch,
        Err(error) => return refuse(RefusalCode::Internal, &error),
    };
    if !switch.is_on() {
        return refuse(
            RefusalCode::ServiceOff,
            &format!("{service} is switched off with this machine on the answering side"),
        );
    }
    let known = match store.peer(peer) {
        Ok(Some(known)) => known,
        Ok(None) => return refuse(RefusalCode::Internal, "the asking machine is not a paired peer here"),
        Err(error) => return refuse(RefusalCode::Internal, &error),
    };
    let registration = match machine.plugins().lookup(&service) {
        crate::plugin::Lookup::Offered(registration) => registration,
        crate::plugin::Lookup::Gone => {
            return refuse(
                RefusalCode::PluginAbsent,
                &format!("the {} plugin that offers {service} here has stopped", service.plugin()),
            )
        }
        crate::plugin::Lookup::NotOffered => {
            return refuse(
                RefusalCode::NoSuchService,
                &format!("no plugin here offers {service}"),
            )
        }
    };
    let call = wsb::plugin::Call {
        service: service.clone(),
        caller: wsb::plugin::Caller {
            id: known.id.to_string(),
            name: known.name.clone(),
            fingerprint: known.fingerprint.as_str().to_string(),
        },
        body,
    };
    match crate::plugin::PluginRegistry::call(&registration, &call) {
        Ok(wsb::plugin::Answer::Reply { body }) => Response::Plugin { body },
        Ok(wsb::plugin::Answer::Refused { detail }) => refuse(RefusalCode::PluginRefused, &detail),
        Err(failure) => {
            tracing::warn!(%peer, %service, failure = %failure.0, "a plugin did not answer a peer's request");
            refuse(RefusalCode::PluginAbsent, &failure.0)
        }
    }
}

/// Put this machine's baseline for one repository back to the commit both
/// machines share.
///
/// The same recovery the asking side performs, done here so a disagreement
/// ends in one round instead of each side waiting for the other to notice.
fn reseed_from_head(
    machine: &dyn Machine,
    peer: &PeerId,
    local: &LocalWorkspace,
    repository: &wsb::peer::model::RepositoryId,
) -> Result<(), String> {
    let repo = local
        .repository(repository)
        .ok_or_else(|| format!("{repository} vanished while re-seeding"))?;
    let seeded = wsb::peer::seed::from_head(&repo.workdir)?;
    let store = machine.store();
    store.put_baseline(peer, &local.id, repository, &seeded)?;
    store.mark_seeded(peer, &local.id, repository, crate::peer::service::now())?;
    tracing::warn!(
        %peer,
        repository = %repository,
        paths = seeded.len(),
        "a peer planned against a different baseline; re-seeded from the commit they share"
    );
    Ok(())
}

fn refuse(code: RefusalCode, detail: &str) -> Response {
    Response::Refused {
        code,
        detail: detail.to_string(),
    }
}

#[cfg(test)]
#[path = "connection_tests.rs"]
mod tests;