monochromatic-git-policy-cli 0.6.0

//! What:
//!  The resident service's loop: accept connections, queue requests in arrival order, land what is queued as a batch,
//!  answer each client, and exit when told to retire.
//! Why:
//!  One writer lands every commit, so there is no lock to take, hold or wait for: order is the order of the queue, the
//!  cost of the one durable write is shared by the batch, and a client that waits blocks on its socket instead of
//!  polling (`doc/planning/cli-git-resident-service.md`). The service never starts itself: the operating system hands it
//!  the listening socket (socket activation), or a development run binds one.
//!
//! In TS you'd write (pseudocode):
//! ```ts
//! for await (const batch of queue.batches()) { const outcomes = await landBatch(batch); for (...) reply(outcome); }
//! ```

pub use crate::service_server_work_loop::work_loop;
pub use crate::service_server_socket_binding::{Bind, INHERITED_DESCRIPTOR, bind, socket_passed_in};
#[cfg(test)]
use crate::service_server_socket_binding::ensure_socket_passed_in;
use crate::service_server_queue_types_and_send::{GroupKey, Job, Queued, id_of_queued, index_of_queued, place_in_groups, send};
use crate::service_server_land_group::{land_group, settle_open_batches};
use crate::service_server_connection_handling::handle_connection;
use crate::service_journal::{DONE_RETAINED, landed_commits, prune_done};
use crate::service_index_lock::unsettled_locks;
use crate::service_index_prestage::{PrestageGuard, start as start_prestage};
use crate::service_captures::CaptureMemory;
use crate::service_land::LandContext;
use crate::service_timing::{ExitReason, exit_line, log};
use crate::private_storage::remove_tree;
use crate::service_claim_watch::{ClaimWatch, WATCH_DIRECTORY};
use crate::push_retry::{PLAN_DIRECTORY, PushRetries};
use crate::service_work::{GitHandlers, Wake};
use crate::service_busy::{GroupAdmission, LandingBoard};
use crate::wait_ceilings::NEVER_ENDED_STEP_SLOW_AFTER;
use crate::service_protocol::{
    FailureCode, PROTOCOL_VERSION, RELEASE, ServiceMessage,
};
use std::collections::HashMap;
use std::os::unix::net::{UnixListener, UnixStream};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel};
use std::sync::Arc;
use std::path::PathBuf;
use std::time::{Duration, Instant};

/// The largest batch whose landing still costs about what its size predicts: 256. Measured with the service binary on throwaway
///  repositories, 2026-11-03, bursts of simultaneous commit requests (`MAX_BATCH` raised to 4096 for the probe, three rounds
///  alternating over 64, 128, 256, 512 and 1024 clients, other sessions running, a debug build): a landing took 71 to 76 ms for
///  62 to 63 requests, 82 to 85 ms for 126, 104 to 109 ms for 253 to 255, 193 to 204 ms for 509 to 511 (once 2,002 ms, a host stall)
///  and 525 to 547 ms for 1,021 to 1,023. The cost of each added request is 0.16 ms up to 254 requests, 0.36 ms from there to
///  510, and 0.66 ms to 1,022, so the cost per request has not yet doubled at 256. The time a client waits for a burst of 128 was
///  176 to 206 ms in one landing and 243 to 249 ms in three at the old cap of 64.
pub const BATCH_COST_KNEE: usize = 256;

/// The most requests landed as one batch: `BATCH_COST_KNEE`. A batch is answered all at once and holds the index lock while it
///  lands, so the cap bounds both. The prototype's burst reached 64, and production never reached more than one request in a
///  batch (1,302 answered requests, 2026-11-02T09:01:38Z to 2026-11-03T15:11:58Z), so the cap bites only in a burst. Option B of the
///  batch size question, picked by the coordinator under the owner's delegation, open to the owner's veto
///  (`doc/planning/cli-git-shipped-limits-audit.md`, "Picked for these follow-ups").
pub const MAX_BATCH: usize = BATCH_COST_KNEE;

/// How long the service waits with nothing to do before it exits; the manager's socket stays, so the next connection starts it
///  again. Bazel's server does the same with `--max_idle_secs`, whose default is three hours, which is this value's precedent.
///  Measured cost of exceeding it: of 1,051 commits in the commit timing log (2026-11-02T09:01:37Z to 2026-11-03T15:10:12Z), the 10
///  that followed an idle gap of more than three hours reached the service in at most 70 ms (`reachMs`) after a start wait of at
///  most 47 ms (`startWaitMs`); 110 idle exits happened in that window. The 28 running services held 212 MiB together (7.9 MiB each,
///  `systemctl --user show --property=MemoryCurrent`). A shorter exit would trade at most that start for that memory; the owner's
///  bound is that a commit should not take 5 s longer than native Git (`SPEC.md`). Kept at three hours as the coordinator's pick under the owner's delegation, open to the owner's veto (`doc/planning/cli-git-shipped-limits-audit.md`, "Sizes and ages of the service's own housekeeping").
pub const IDLE_EXIT: Duration = Duration::from_secs(3 * 60 * 60);

/// How a service run behaves apart from where it listens and what it lands.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ServeSettings {
    /// How long the service waits with nothing to do before it exits.
    pub idle: Duration,
    /// The release the service announces to every client. It is this build's release (`RELEASE`) except in a test that
    ///  needs a service of another release.
    pub release: String,
    /// The worktree this instance serves, as `service_instance::instance_of` names it; every message about another worktree
    ///  is refused. Nothing in a development run that serves any worktree.
    pub instance: Option<String>,
    /// How long a write the service never ends runs before the clients behind it are told, and may be withdrawn while the
    ///  service is inside it (`wait_ceilings.rs`, `NEVER_ENDED_STEP_SLOW_AFTER`); a control shortens it.
    pub stuck_after: Duration,
}

impl ServeSettings {
    /// The settings of an ordinary run: the given idle exit and this build's release.
    pub fn new(idle: Duration) -> Self {
        return Self { idle, release: String::from(RELEASE), instance: None, stuck_after: NEVER_ENDED_STEP_SLOW_AFTER };
    }
}

/// Starts a thread the service cannot do without, or ends the service with a plain line and status 4: a service that goes on
///  without its acceptor or its work thread answers no one, while an ended service is started again by the manager.
fn start_thread_or_exit(what: &str, body: impl FnOnce() + Send + 'static) {
    if let Err(error) = std::thread::Builder::new().spawn(body) {
        crate::service_error!("cannot start the thread for {what}: {error}");
        std::process::exit(4);
    }
}

/// How many withdrawn identities the service remembers before it forgets them all: the same count as the finished batches it
///  remembers (`DONE_RETAINED`), since a client that withdrew sends nothing more for its request.
const WITHDRAWN_RETAINED: usize = DONE_RETAINED;

/// How long the acceptor waits after a connection it could not give a thread.
const ACCEPT_BACKOFF: Duration = Duration::from_millis(100);

/// What:
///  Runs the service until it is told to retire.
/// Why:
///   Recovery runs before the first connection is accepted; then the loop takes everything queued as one batch, groups it
///  by branch in order of first arrival, lands each group, and answers.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// await serve(listener, context)
/// ```
pub fn serve(listener: UnixListener, context: &LandContext, settings: &ServeSettings) {
    let serving_since: Instant = Instant::now();
    let idle: Duration = settings.idle;
    let _reports = settle_open_batches(context, settings.instance.as_deref());
    // The memory of landed requests is bounded: the oldest finished batches go before the rest are read.
    let pruned: usize = prune_done(context.journal.as_path(), DONE_RETAINED);
    if pruned > 0 {
        crate::service_log!("removed the records of {pruned} old finished batches from {}", context.journal.display());
    }
    let mut landed: HashMap<String, String> = landed_commits(context.journal.as_path());
    // What the earlier release's helper of this worktree finished counts as landed too: a client it never answered asks again.
    if let Some(instance) = settings.instance.as_deref() {
        landed.extend(crate::earlier_service_journal_adopted::earlier_commits_landed(context.journal.as_path(), instance));
    }
    // The identities of the requests their clients withdrew; a request with one of these is never landed.
    let mut withdrawn: std::collections::HashSet<String> = std::collections::HashSet::new();
    // What the landed commits were captured at: owned by this loop, the only thread that lands and answers.
    let mut captures: CaptureMemory = CaptureMemory::default();
    let (sender, receiver): (Sender<Job>, Receiver<Job>) = channel();
    let acceptor_queue: Sender<Job> = sender.clone();
    // After-commit work (push, upkeep) runs on its own thread, so a slow network never delays a landing.
    let (work_sender, work_receiver): (Sender<Wake>, Receiver<Wake>) = channel();
    let handlers: GitHandlers = GitHandlers::new(context.git.clone());
    let busy: Arc<AtomicUsize> = Arc::new(AtomicUsize::new(0));
    let worker_busy: Arc<AtomicUsize> = Arc::clone(&busy);
    // What this landing thread is inside, for the threads that talk to clients (`service_busy.rs`).
    let board: Arc<LandingBoard> = LandingBoard::new(context.journal.clone(), settings.stuck_after);
    crate::service_busy_step_record_on_this_thread::use_on_this_thread(Arc::clone(&board));
    let connection_board: Arc<LandingBoard> = Arc::clone(&board);
    // The plan of the pushes waiting for a new try is kept in this instance's own journal, so a restarted service rebuilds it.
    let plan_directory: PathBuf = context.journal.join(PLAN_DIRECTORY);
    // The repositories this instance was woken for are kept in the journal the same way, so a restarted service looks at them.
    let watch_directory: PathBuf = context.journal.join(WATCH_DIRECTORY);
    let watch_instance: Option<String> = settings.instance.clone();
    start_thread_or_exit("the after-commit work", move || { work_loop(&work_receiver, &handlers, &worker_busy, PushRetries::restored_from_environment(plan_directory.as_path()), ClaimWatch::restored(watch_directory.as_path(), watch_instance.as_deref())); });
    let announced_release: String = settings.release.clone();
    let connection_busy: Arc<AtomicUsize> = Arc::clone(&busy);
    let instance: Option<String> = settings.instance.clone();
    start_thread_or_exit("the acceptor of connections", move || {
        for connection in listener.incoming().flatten() {
            let queue: Sender<Job> = acceptor_queue.clone();
            let work: Sender<Wake> = work_sender.clone();
            let release: String = announced_release.clone();
            let counted: Arc<AtomicUsize> = Arc::clone(&connection_busy);
            let record: Arc<LandingBoard> = Arc::clone(&connection_board);
            let mine: Option<String> = instance.clone();
            // A connection the service cannot give a thread is told so and closed; the acceptor goes on, because a panic here
            //  would end the acceptor and leave a service that is running and answers no one.
            let notice: Option<UnixStream> = connection.try_clone().ok();
            let spawned = std::thread::Builder::new().spawn(move || { handle_connection(connection, &queue, &work, &counted, &record, release.as_str(), mine.as_deref()); });
            if let Err(error) = spawned {
                crate::service_error!("no thread for a connection, which is closed after telling the client: {error}");
                if let Some(stream) = notice {
                    send(&stream, &ServiceMessage::Welcome { version: PROTOCOL_VERSION, release: Some(announced_release.clone()) });
                    send(&stream, &ServiceMessage::Failed { id: String::new(), code: FailureCode::Unavailable });
                }
                // Starting a thread keeps failing while the pressure lasts; waiting a moment keeps the acceptor from spinning on it.
                std::thread::sleep(ACCEPT_BACKOFF);
            }
        }
    });
    // Microseconds, so that instances started at different times count from different numbers; the number each batch takes is
    //  still claimed in the journal (`claim_batch`), because equal readings and counts that overtake another's are possible.
    let clock_seed: u64 = match std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) {
        Ok(elapsed) => u64::try_from(elapsed.as_micros()).unwrap_or(1),
        Err(_before_the_epoch) => 1,
    };
    let mut seq: u64 = crate::service_fault::pinned_first_batch().unwrap_or(clock_seed);
    // The marker made ahead is removed whichever way this loop ends.
    let _prestaged = PrestageGuard;
    let mut retiring: bool = false;
    while !retiring {
        let first: Job = match receiver.recv_timeout(idle) {
            Ok(job) => job,
            // Work in flight keeps the service; otherwise it exits and the manager starts it on the next connection.
            Err(RecvTimeoutError::Timeout) if busy.load(Ordering::SeqCst) > 0 => continue,
            Err(RecvTimeoutError::Timeout) => {
                // Leaving: the syncs of the landings that were answered are made before the service goes (a signal cannot wait for them).
                crate::service_durable::drain();
                log(exit_line(ExitReason::Idle, serving_since.elapsed()).as_str());
                return;
            }
            Err(RecvTimeoutError::Disconnected) => {
                // Leaving: the syncs of the landings that were answered are made before the service goes (a signal cannot wait for them).
                crate::service_durable::drain();
                log(exit_line(ExitReason::Error, serving_since.elapsed()).as_str());
                return;
            }
        };
        let mut jobs: Vec<Job> = vec![first];
        while jobs.len() < MAX_BATCH {
            match receiver.try_recv() {
                Ok(job) => jobs.push(job),
                Err(_) => break,
            }
        }
        // Withdrawals are answered before anything lands, on this one thread, so each is either after a landing that is over (the
        //  commit is on the branch, and that is the answer) or before the landing of its request begins (the request is dropped
        //  and remembered, so it can never land, not even when it reaches the queue after the withdrawal).
        if withdrawn.len() > WITHDRAWN_RETAINED {
            withdrawn.clear();
        }
        for job in &jobs {
            if let Job::Withdraw(id, reply) = job {
                match landed.get(id) {
                    Some(commit) => reply.send(&ServiceMessage::Landed { id: id.clone(), commit: commit.clone(), already: true }),
                    None => {
                        withdrawn.insert(id.clone());
                        reply.send(&ServiceMessage::Withdrawn { id: id.clone() });
                    }
                }
            }
        }
        // Branches in order of first arrival; within a branch, arrival order.
        let mut groups: Vec<(GroupKey, Vec<Queued>)> = Vec::new();
        let mut releases: Vec<std::path::PathBuf> = Vec::new();
        for job in jobs {
            match job {
                Job::Withdraw(..) => {}
                Job::Commit(request, _dropped, _arrival) if withdrawn.contains(&request.id) => {
                    crate::service_log!("dropped the request {} that its client withdrew", request.id);
                }
                Job::Retire => retiring = true,
                Job::Activity => {}
                Job::Release(path) => releases.push(path),
                Job::Commit(request, stream, arrival) => {
                    // An amend is a group of its own, and what arrives after it starts another group, so the order on
                    //  one branch is the order of arrival.
                    // `HEAD` is a head of one worktree, so two detached worktrees of one repository are two branches.
                    let queue_reference: String = if request.reference == "HEAD" {
                        format!("HEAD@{}", request.repository.display())
                    } else {
                        request.reference.clone()
                    };
                    let key: GroupKey = (request.common_dir.clone(), queue_reference, request.lands_alone());
                    place_in_groups(&mut groups, key, (request, stream, arrival));
                }
            }
        }
        for (_, queued_group) in groups {
            let index: Option<std::path::PathBuf> = queued_group.first().map(index_of_queued);
            // A request whose client was answered "withdrawn" while this thread was stuck in a write is dropped here, and the
            //  rest are marked as being landed, under one lock (`service_busy.rs`).
            let ids: Vec<String> = queued_group.iter().map(id_of_queued).collect();
            let admission: GroupAdmission = board.begin_group(ids.as_slice());
            for id in &admission.dropped {
                crate::service_log!("dropped the request {id} that its client withdrew while the service was stuck in a write");
            }
            for id in &admission.unavailable {
                crate::service_error!("the request {id} could not be admitted to its group (the record of the landing thread cannot be written), so it is answered with a failure and not landed");
            }
            // A dropped request is not answered here; an unavailable one is answered with a failure; the rest are landed.
            let mut unavailable: Vec<Queued> = Vec::new();
            let mut admitted: Vec<Queued> = Vec::new();
            for queued in queued_group {
                if admission.dropped.contains(&queued.0.id) {
                    continue;
                }
                if admission.unavailable.contains(&queued.0.id) {
                    unavailable.push(queued);
                } else {
                    admitted.push(queued);
                }
            }
            for (request, stream, _) in &unavailable {
                stream.send(&ServiceMessage::Failed { id: request.id.clone(), code: FailureCode::Unavailable });
            }
            let group: Vec<Queued> = admitted;
            if group.is_empty() {
                board.end_group(ids.as_slice());
                continue;
            }
            land_group(context, settings.instance.as_deref(), &mut seq, group, &mut landed, &mut captures);
            board.end_group(ids.as_slice());
            // The batch after this one is numbered one more and lands on this index (a worktree has one), so its marker is
            //  made and flushed now, while the service is idle, and the next commit only places it.
            if let Some(next_index) = index {
                start_prestage(next_index.as_path(), seq + 1);
            }
        }
        // A lock this thread could not settle names this process: the restart reclaims it once this process is gone.
        if unsettled_locks() > 0 {
            crate::service_error!("ending, because an index lock could not be settled; the next connection starts the service again and its recovery reclaims the lock");
            return;
        }
        for path in releases {
            if let Err(error) = remove_tree(path.as_path()) {
                crate::service_warn!("the scratch directory {} could not be removed: {error}", path.display());
            }
        }
    }
    // Leaving: the syncs of the landings that were answered are made before the service goes (a signal cannot wait for them).
    crate::service_durable::drain();
    log(exit_line(ExitReason::Retire, serving_since.elapsed()).as_str());
}

/// Socket controls stay out of the release executable.
#[cfg(test)]
#[path = "service_server_tests.rs"]
mod tests;