gunnar-grpc 1.5.0

//! `Mirror` pull-mirrors a remote forge into a gunnar repo (mirroring).
//!
//! A successful sync publishes `MirrorSynced` (the event stream), which is also the hook
//! replication will hang off. `Sync` itself needs `gunnar-client` and
//! a real store, so the seam answers `UNIMPLEMENTED` until both land. Never
//! "0 refs changed", which would look exactly like a mirror that is up to date.

use gunnar_proto::mirror_server::Mirror;
use gunnar_proto::{
    LogChunk, LogFetchRequest, LogHeadRequest, LogHeadResponse,
    AbdicateReplicationRequest, AddMirrorRequest, AddMirrorResponse, DemoteReplicationRequest,
    FlushReplicationRequest, FlushReplicationResponse, MirrorCredentialKind, MirrorStatusRequest, MirrorStatusResponse, RemoveMirrorRequest,
    RemoveMirrorResponse, RemoveReplicationRequest, ReplicationRole, ReplicationStatusRequest,
    ReplicationStatusResponse, ResyncReplicationRequest, ResyncReplicationResponse,
    SetMirrorCredentialRequest, SetMirrorCredentialResponse, SetReplicationRequest, SetRoleRequest,
    RepoPath, SyncMirrorRequest, SyncMirrorResponse,
};
use tonic::{Request, Response, Status};

use crate::auth::Access;
use crate::service::MirrorService;
use crate::status::{opt_repo_path, opt_str, repo_path};
use gunnar_proto::GrantLevel;

/// `Mirror.SnapshotFetch`: pieces queued between the producer and the stream at most — the
/// primary's memory for a snapshot of any size is this many
/// [`LOG_PIECE_BYTES`](crate::seam::LOG_PIECE_BYTES) pieces.
const SNAPSHOT_WINDOW: usize = 4;

/// `Mirror.LogFetch`: the same window (twin-streaming, 2026-10-06) — the primary's memory for a
/// range, whatever its entries' sizes, is this many pieces.
const LOG_FETCH_WINDOW: usize = 4;

impl MirrorService {
    /// **The scope a verb on one mirror is asked at.** The repository — or,
    /// for an import whose repository is still to come (`repo_pending`: the
    /// first sync makes it, 2026-10-18), the namespace, which is exactly what
    /// `Mirror.Add { create_repo }` was asked at. A repository-scoped grant
    /// cannot be evaluated against a row that does not exist yet, so without
    /// this the person who started the import could not set its token, sync
    /// it or remove it.
    fn mirror_scope(&self, path: &gunnar_proto::RepoPath) -> Access {
        if self.ctx.control.get_repo(path).is_err()
            && matches!(
                self.ctx.control.mirror_status(None, Some(path)),
                Ok(mirrors) if mirrors.iter().any(|m| m.repo_pending)
            )
        {
            return Access::namespace(&path.namespace);
        }
        Access::repo(path)
    }

    /// **Who reads the transaction log**: a member of group
    /// [`crate::TWIN_GROUP`] (the twin's own machine key — it holds the
    /// log, it is not an administrator of the box), or a principal with the
    /// server grant at `level`. Everyone else is refused exactly as `require`
    /// refuses them.
    fn require_log_reader(&self, caller: &crate::auth::Caller, level: GrantLevel, method: &'static str) -> Result<(), Status> {
        if self.ctx.control.is_group_member(&caller.principal_id, crate::TWIN_GROUP).unwrap_or(false) {
            tracing::info!(
                target: crate::auth::AUDIT_TARGET,
                event = "grpc.call",
                method,
                principal = %caller.principal_id,
                fingerprint = %caller.fingerprint,
                decision = "allow",
                "a member of group twin reads the transaction log"
            );
            return Ok(());
        }
        self.ctx.require(caller, Access::Server, level, method)
    }

    /// ★ **A twin owed a growth does not follow** (cov-13, 2026-10-06): main's set grew
    /// while it was silent and its last report (this poll's `LogHead`) holds less — it would
    /// fill part-way through main's data. Refused by name ([`crate::TWIN_TOO_SMALL`]) before
    /// anything is read; the twin waits in `recovering`, polling, and is grown.
    fn refuse_undersized(&self, caller: &crate::auth::Caller, method: &'static str) -> Result<(), Status> {
        match self.ctx.heartbeats.follow_refusal(&caller.principal_id) {
            None => Ok(()),
            Some(why) => {
                tracing::info!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method, principal = %caller.principal_id, decision = "deny", reason = %why, "a twin owed a growth may not follow yet");
                Err(Status::failed_precondition(why))
            }
        }
    }
}

#[tonic::async_trait]
impl Mirror for MirrorService {
    type LogFetchStream = std::pin::Pin<
        Box<dyn tokio_stream::Stream<Item = Result<LogChunk, Status>> + Send + 'static>,
    >;

    async fn log_head(
        &self,
        request: Request<LogHeadRequest>,
    ) -> Result<Response<LogHeadResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.require_log_reader(&caller, GrantLevel::Read, "Mirror.LogHead")?;
        let address = request.remote_addr().map(|a| a.to_string()).unwrap_or_default();
        let request = request.into_inner();
        let report = request.report;
        // A twin's proof that a newer generation of this log exists (stinky row
        // 18): an old main that finds it genuine fences itself before answering.
        if let Some(proof) = request.generation_proof.as_ref() {
            self.ctx.control.fence_by_generation(proof)?;
        }
        // The caller's freshness challenge; the answer is signed over it. Empty
        // from anything that is not a twin's poll, and then it is unsigned.
        let head = self.ctx.control.log_head_signed(&request.nonce)?;
        // The dual heartbeat: a twin's poll is recorded against the head it
        // was answered with, so the lag is the one this answer creates.
        if let Some(report) = report {
            self.ctx.heartbeats.record(&caller.principal_id, &address, &report, head.head_seq, estate_clock::now_unix_ms());
        }
        Ok(Response::new(head))
    }

    /// Server Admin only (never group `twin`): retention decides what every twin must hold.
    async fn log_retain(
        &self,
        request: Request<gunnar_proto::LogRetainRequest>,
    ) -> Result<Response<LogHeadResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx.require(&caller, Access::Server, GrantLevel::Admin, "Mirror.LogRetain")?;
        let gunnar_proto::LogRetainRequest { from, allow_past_twin } = request.into_inner();
        // **The retention guard** (2026-09-22, the owner's ruling). Retaining from `from` drops
        // every frame below it; the twin that has applied least needs `applied + 1` onward.
        // Past that, refused by name with both numbers — unless the caller says
        // `allow_past_twin`, and then the trail records who overrode which twin.
        let wanted = if from == 0 { self.ctx.control.log_head()?.head_seq + 1 } else { from };
        // **An enrolled twin that has not reported since this start** (2026-09-22, the owner's
        // ruling): the registry is in memory, so right after a primary restart it knows no
        // position at all — and "unguarded" then would be the one window the guard exists for.
        // Membership is read the way `Admin.Health` reads it (`service/admin.rs` `twin_rows`:
        // `list_group_members(TWIN_GROUP)`); a plane with no such group enrols nobody.
        let silent: Vec<String> = self
            .ctx
            .control
            .list_group_members(crate::TWIN_GROUP)
            .map(|ps| ps.into_iter().map(|p| p.id.to_string()).filter(|id| !self.ctx.heartbeats.has_reported(id)).collect())
            .unwrap_or_default();
        if !silent.is_empty() {
            let why = format!(
                "retaining from {wanted} while enrolled twin(s) {} have reported no position since this \
                 server started: what they have applied is unknown, so nothing says these frames are \
                 not the ones they need",
                silent.iter().map(|t| format!("'{t}'")).collect::<Vec<_>>().join(", ")
            );
            if !allow_past_twin {
                crate::auth::audit_denied("Mirror.LogRetain", &caller, &Access::Server, GrantLevel::Admin, self.ctx.config.auth, &why, self.ctx.ledger());
                return Err(Status::failed_precondition(format!(
                    "retention-twin-position-unknown: {why}. Wait for the twin's next poll (every few \
                     seconds while it follows), or pass allow_past_twin (`gunnar control twin retain \
                     --allow-past-twin`) if the twin is gone or the hole is the point"
                )));
            }
            crate::auth::audit_override("Mirror.LogRetain", &caller, &Access::Server, GrantLevel::Admin, self.ctx.config.auth, &why, self.ctx.ledger());
        }
        match self.ctx.heartbeats.lowest_applied() {
            Some((twin, applied)) if wanted > applied + 1 => {
                let why = format!(
                    "retaining from {wanted} drops frames {}..={} that twin '{twin}' has not applied \
                     (it reported applied {applied}); it would have to take a snapshot to close the hole",
                    applied + 1,
                    wanted - 1
                );
                if !allow_past_twin {
                    crate::auth::audit_denied("Mirror.LogRetain", &caller, &Access::Server, GrantLevel::Admin, self.ctx.config.auth, &why, self.ctx.ledger());
                    return Err(Status::failed_precondition(format!(
                        "retention-past-twin: {why}. Retain from {} or lower, let the twin catch up, or \
                         pass allow_past_twin (`gunnar control twin retain --allow-past-twin`) if the \
                         twin is gone or the hole is the point",
                        applied + 1
                    )));
                }
                crate::auth::audit_override("Mirror.LogRetain", &caller, &Access::Server, GrantLevel::Admin, self.ctx.config.auth, &why, self.ctx.ledger());
            }
            Some(_) => {}
            None if silent.is_empty() => tracing::warn!(
                target: crate::auth::AUDIT_TARGET,
                method = "Mirror.LogRetain",
                from = wanted,
                "no twin is enrolled and none has polled since this server started; retention is not guarded"
            ),
            // Enrolled but silent: refused or overridden (and audited) above.
            None => {}
        }
        let head = self.ctx.control.log_retain(from)?;
        tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = "Mirror.LogRetain", principal = %caller.principal_id, from, first = head.first_seq, decision = "allow", "the transaction log's retention moved");
        Ok(Response::new(head))
    }

    /// Admin: the log carries every sealed secret's bytes and every pack.
    async fn log_fetch(
        &self,
        request: Request<LogFetchRequest>,
    ) -> Result<Response<Self::LogFetchStream>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.require_log_reader(&caller, GrantLevel::Admin, "Mirror.LogFetch")?;
        self.refuse_undersized(&caller, "Mirror.LogFetch")?;
        let request = request.into_inner();
        if request.from == 0 {
            return Err(Status::invalid_argument(
                "from is required: sequence numbers start at 1",
            ));
        }
        // **Streamed, never gathered** (twin-streaming, 2026-10-06). This answered
        // `tokio_stream::iter(chunks_of(entries))`: the whole range read into memory, then cut
        // into a second whole copy of 1 MiB messages before the first one left — a 256 MiB pack
        // cost the primary 512 MiB per fetch. The refusals are still answered here, as the
        // call's status; the range is then read a piece at a time on a blocking thread into a
        // channel of LOG_FETCH_WINDOW pieces that tonic drains as the twin's HTTP/2 window
        // opens. A reader that stops taking pieces is cut off ([`LOG_FETCH_STALL_TIMEOUT`]).
        // `from_offset` (restore-large-entry, 2026-10-07): a restore resuming an entry the box
        // being filled already holds part of starts that entry at the byte it reached.
        let produce = self
            .ctx
            .control
            .log_fetch_stream_at(request.from, request.from_offset, request.to, request.max_bytes)?;
        let (tx, rx) = tokio::sync::mpsc::channel::<Result<LogChunk, Status>>(LOG_FETCH_WINDOW);
        let rt = tokio::runtime::Handle::current();
        let (from, to) = (request.from, request.to);
        tokio::task::spawn_blocking(move || {
            let mut sink = |chunk: LogChunk| send_log_piece(&rt, &tx, chunk, LOG_FETCH_STALL_TIMEOUT);
            if let Err(err) = produce(&mut sink) {
                tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = "Mirror.LogFetch", from, to, reason = %err, "the log stream stopped");
                let _ = tx.blocking_send(Err(Status::from(err)));
            }
        });
        Ok(Response::new(Box::pin(tokio_stream::wrappers::ReceiverStream::new(rx))))
    }

    type SnapshotFetchStream = std::pin::Pin<
        Box<dyn tokio_stream::Stream<Item = Result<LogChunk, Status>> + Send + 'static>,
    >;

    /// Admin or group `twin`: a snapshot carries every sealed secret and every object.
    async fn snapshot_fetch(
        &self,
        request: Request<gunnar_proto::SnapshotFetchRequest>,
    ) -> Result<Response<Self::SnapshotFetchStream>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.require_log_reader(&caller, GrantLevel::Admin, "Mirror.SnapshotFetch")?;
        self.refuse_undersized(&caller, "Mirror.SnapshotFetch")?;
        // From where the asker holds it (restore-after-snapshot): a twin's base resumed.
        let asked = request.into_inner();
        // Streamed, never gathered: the producer runs on a blocking thread and hands each piece
        // to a channel of SNAPSHOT_WINDOW pieces; tonic takes one when the twin's HTTP/2 window
        // has room. A slow twin holds the producer back, so the primary never holds more than
        // the window (plus the producer's one buffer and tonic's send buffer). A twin that
        // STOPS taking pieces is cut off by [`SNAPSHOT_STALL_TIMEOUT`] — the producer is
        // holding the log's freeze and this primary's writes with it.
        let (tx, rx) = tokio::sync::mpsc::channel::<Result<LogChunk, Status>>(SNAPSHOT_WINDOW);
        let control = std::sync::Arc::clone(&self.ctx.control);
        let rt = tokio::runtime::Handle::current();
        tokio::task::spawn_blocking(move || {
            let mut sink = |chunk: LogChunk| send_piece(&rt, &tx, chunk, SNAPSHOT_STALL_TIMEOUT);
            if let Err(err) = control.snapshot_stream_from(asked.from_item, asked.from_offset, &asked.expect_id, &mut sink) {
                tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = "Mirror.SnapshotFetch", reason = %err, "the snapshot stream stopped");
                let _ = tx.blocking_send(Err(Status::from(err)));
            }
        });
        Ok(Response::new(Box::pin(tokio_stream::wrappers::ReceiverStream::new(rx))))
    }

    async fn add(
        &self,
        request: Request<AddMirrorRequest>,
    ) -> Result<Response<AddMirrorResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        let path = repo_path(request.repo, "repo")?;
        // Pointing a repository at a remote decides what lands in it. Admin,
        // and namespace-scoped when the repository is to be created, because
        // there is nothing narrower yet.
        let access = if request.create_repo {
            Access::namespace(&path.namespace)
        } else {
            Access::repo(&path)
        };
        self.ctx
            .require(&caller, access, GrantLevel::Admin, "Mirror.Add")?;
        // **An encrypted repository is never a mirror** (wiring tracer B
        // 2026-10-08): a mirror's fetched pack is an upstream's plaintext and
        // lands with no encrypted gate, framed for the twin. Asked after the
        // grant check, as the collision below is.
        if matches!(self.ctx.control.get_repo(&path), Ok(repo) if repo.encrypted) {
            return Err(Status::failed_precondition(encrypted_is_never_a_mirror(&path)));
        }
        // **A mirror that would create its repository never takes over one
        // that exists** (2026-10-04). An import whose name collided used to
        // turn somebody's repository into a mirror, and a pruning sync then
        // moved its branches to the upstream's. Re-adding an existing MIRROR
        // is unchanged; `adopt` takes a plain repository over on purpose.
        // Asked after the grant check, so the refusal is no existence oracle
        // for a caller who could not add here anyway.
        if request.create_repo && !request.adopt && self.ctx.control.get_repo(&path).is_ok() {
            let already_a_mirror = match self.ctx.control.mirror_status(None, Some(&path)) {
                Ok(mirrors) => !mirrors.is_empty(),
                Err(crate::seam::ControlError::NotFound(_))
                | Err(crate::seam::ControlError::Unimplemented(_)) => false,
                Err(err) => return Err(err.into()),
            };
            if !already_a_mirror {
                return Err(Status::failed_precondition(format!(
                    "{}/{} already exists and is not a mirror, so it is not taken over: \
                     import under another name, or say `adopt` to make this repository a \
                     mirror of {} (its next sync moves its refs to the upstream's)",
                    path.namespace, path.repo, request.remote_url
                )));
            }
        }
        let mirror = self.ctx.control.mirror_add(
            &path,
            &request.remote_url,
            request.prune,
            request.create_repo,
        )?;
        Ok(Response::new(AddMirrorResponse {
            mirror: Some(mirror),
        }))
    }

    /// **Enqueue, and answer with the job.**
    ///
    /// A fetch is minutes of somebody else's network; a unary RPC that waited
    /// for one would time out on exactly the repositories worth mirroring. So
    /// this returns as soon as the sync is on the queue, and the caller polls
    /// `Mirror.Status` for `job.state`.
    ///
    /// `refs_changed` and `updates` are therefore 0 and empty here, and
    /// `SyncMirrorResponse`'s own comment says so on the wire: they are filled
    /// in on the JOB, once it is DONE. The `MirrorSynced` event (the event stream) is
    /// published by the queue's worker when the sync actually completes — from
    /// here it would be an event announcing something that has not happened.
    async fn sync(
        &self,
        request: Request<SyncMirrorRequest>,
    ) -> Result<Response<SyncMirrorResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let path = repo_path(request.into_inner().repo, "repo")?;
        // A sync moves refs, so it is at least a push: Write.
        self.ctx.require(
            &caller,
            self.mirror_scope(&path),
            GrantLevel::Write,
            "Mirror.Sync",
        )?;
        // A registration made before the door refused it, or a flag set below
        // the door: never fetched into (the worker refuses it too).
        if matches!(self.ctx.control.get_repo(&path), Ok(repo) if repo.encrypted) {
            return Err(Status::failed_precondition(encrypted_is_never_a_mirror(&path)));
        }
        let mirror = self.ctx.control.mirror_sync(&path)?;
        let job = mirror.job.clone();
        Ok(Response::new(SyncMirrorResponse {
            mirror: Some(mirror),
            refs_changed: 0,
            updates: Vec::new(),
            job,
        }))
    }

    /// **Write the credential; never read it back.**
    ///
    /// `Admin` on the repository, the same level `Mirror.Add` takes: pointing a
    /// repository at a remote and saying what it authenticates with are one
    /// decision, and neither is a `Write`.
    ///
    /// The request is the only message in this service that carries a secret,
    /// and it carries it INTO the server. Nothing in the response, in
    /// `MirrorConfig`, in `Mirror.Status` or in any error from the sync that
    /// uses it contains the value — `MirrorConfig.credential` says which kind
    /// is stored and that is the whole of what any reader gets.
    async fn set_credential(
        &self,
        request: Request<SetMirrorCredentialRequest>,
    ) -> Result<Response<SetMirrorCredentialResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        let path = repo_path(request.repo, "repo")?;
        self.ctx.require(
            &caller,
            self.mirror_scope(&path),
            GrantLevel::Admin,
            "Mirror.SetCredential",
        )?;
        let kind = MirrorCredentialKind::try_from(request.kind).map_err(|_| {
            Status::invalid_argument(format!(
                "{} is not a MirrorCredentialKind this server knows",
                request.kind
            ))
        })?;
        let mirror = self.ctx.control.mirror_set_credential(
            &path,
            kind,
            &request.secret,
            &request.username,
        )?;
        Ok(Response::new(SetMirrorCredentialResponse {
            mirror: Some(mirror),
        }))
    }

    async fn status(
        &self,
        request: Request<MirrorStatusRequest>,
    ) -> Result<Response<MirrorStatusResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let request = request.into_inner();
        let path = opt_repo_path(request.repo, "repo")?;
        let access = match (&path, opt_str(&request.namespace)) {
            (Some(path), _) => Access::repo(path),
            (None, Some(namespace)) => Access::namespace(namespace),
            (None, None) => Access::Server,
        };
        self.ctx
            .require(&caller, access, GrantLevel::Read, "Mirror.Status")?;
        let mirrors = self
            .ctx
            .control
            .mirror_status(opt_str(&request.namespace), path.as_ref())?;
        Ok(Response::new(MirrorStatusResponse { mirrors }))
    }

    async fn remove(
        &self,
        request: Request<RemoveMirrorRequest>,
    ) -> Result<Response<RemoveMirrorResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let path = repo_path(request.into_inner().repo, "repo")?;
        self.ctx.require(
            &caller,
            self.mirror_scope(&path),
            GrantLevel::Admin,
            "Mirror.Remove",
        )?;
        self.ctx.control.mirror_remove(&path)?;
        Ok(Response::new(RemoveMirrorResponse {}))
    }

    // ── replication: naming a standby on a box with no shell ───────────────
    //
    // `Access::Server` and `GrantLevel::Admin` on all three, INCLUDING the
    // read. A replication target is a URL that every repository on this server
    // is copied to, plus the path of the deploy key it authenticates with; that
    // is server-wide by construction, and reading the roster tells you where a
    // second copy of everything lives. Neither half of that is a per-repository
    // question and neither is a `Read`.

    async fn set_replication(
        &self,
        request: Request<SetReplicationRequest>,
    ) -> Result<Response<ReplicationStatusResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx.require(
            &caller,
            Access::Server,
            GrantLevel::Admin,
            "Mirror.SetReplication",
        )?;
        let target = request
            .into_inner()
            .target
            .ok_or_else(|| Status::invalid_argument("target is required"))?;
        Ok(Response::new(self.ctx.control.replication_set(target)?))
    }

    async fn remove_replication(
        &self,
        request: Request<RemoveReplicationRequest>,
    ) -> Result<Response<ReplicationStatusResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx.require(
            &caller,
            Access::Server,
            GrantLevel::Admin,
            "Mirror.RemoveReplication",
        )?;
        let name = request.into_inner().name;
        if name.trim().is_empty() {
            return Err(Status::invalid_argument(
                "name is required: a standby is removed BY NAME, and an empty name would                  remove whichever one happened to be first",
            ));
        }
        Ok(Response::new(self.ctx.control.replication_remove(&name)?))
    }

    async fn demote_replication(
        &self,
        request: Request<DemoteReplicationRequest>,
    ) -> Result<Response<ReplicationStatusResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx.require(
            &caller,
            Access::Server,
            GrantLevel::Admin,
            "Mirror.DemoteReplication",
        )?;
        // No body to validate: demote clears the whole roster, so there is no
        // name that could be blank and no scope to narrow.
        let _ = request.into_inner();
        Ok(Response::new(self.ctx.control.replication_demote()?))
    }

    /// **Set the role.** `Access::Server` + `Admin`, like the rest of the
    /// family: making a box a standby, or promoting it to primary, decides
    /// whether it takes writes for every repository at once.
    async fn set_role(
        &self,
        request: Request<SetRoleRequest>,
    ) -> Result<Response<ReplicationStatusResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx
            .require(&caller, Access::Server, GrantLevel::Admin, "Mirror.SetRole")?;
        // An unknown enum value is refused by name rather than read as the
        // default `PRIMARY`: with BROWSE in the enum a client one release ahead
        // could otherwise be answered with the one role it must never get.
        let raw = request.into_inner().role;
        let role = ReplicationRole::try_from(raw).map_err(|_| {
            Status::invalid_argument(format!(
                "unknown replication role {raw}; PRIMARY, STANDBY or BROWSE"
            ))
        })?;
        // The repair principal FENCES: it may make a box a standby, never promote one.
        if caller.principal_id == crate::REPAIR_PRINCIPAL && role != ReplicationRole::Standby {
            return Err(Status::permission_denied(format!(
                "principal {} fences only: Mirror.SetRole to {} is not the repair key's",
                caller.principal_id,
                role.as_str_name()
            )));
        }
        Ok(Response::new(self.ctx.control.replication_set_role(role)?))
    }

    /// **Abdicate.** `Access::Server` + `Admin`: the master steps down, which
    /// stops it taking writes and clears its roster — a server-wide act.
    async fn abdicate_replication(
        &self,
        request: Request<AbdicateReplicationRequest>,
    ) -> Result<Response<ReplicationStatusResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx.require(
            &caller,
            Access::Server,
            GrantLevel::Admin,
            "Mirror.AbdicateReplication",
        )?;
        // No body: abdication is a fixed sequence on THIS box.
        let _ = request.into_inner();
        Ok(Response::new(self.ctx.control.replication_abdicate()?))
    }

    async fn replication_status(
        &self,
        request: Request<ReplicationStatusRequest>,
    ) -> Result<Response<ReplicationStatusResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx.require(
            &caller,
            Access::Server,
            GrantLevel::Admin,
            "Mirror.ReplicationStatus",
        )?;
        // `include_refs` rides the request: a plain `replication list` sends
        // false and opens no repository store; `--refs` sends true and pays the
        // one walk the operator asked for.
        let include_refs = request.into_inner().include_refs;
        Ok(Response::new(
            self.ctx.control.replication_status(include_refs)?,
        ))
    }

    /// **Backfill a standby.** `Access::Server` + `Admin`, like the other three:
    /// it moves every repository on this server to another box.
    async fn resync_replication(
        &self,
        request: Request<ResyncReplicationRequest>,
    ) -> Result<Response<ResyncReplicationResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx.require(
            &caller,
            Access::Server,
            GrantLevel::Admin,
            "Mirror.ResyncReplication",
        )?;
        let request = request.into_inner();
        if request.name.trim().is_empty() {
            return Err(Status::invalid_argument(
                "name is required: a standby is backfilled BY NAME, and an empty name would \
                 backfill whichever one happened to be first",
            ));
        }
        let repo = opt_repo_path(request.repo, "repo")?;
        let except = request
            .except
            .into_iter()
            .map(|path| repo_path(Some(path), "except"))
            .collect::<Result<Vec<_>, _>>()?;
        Ok(Response::new(self.ctx.control.replication_resync(
            &request.name,
            repo.as_ref(),
            &except,
        )?))
    }

    /// **The replication watermark** (`DATA-SET-GROWTH-FLOW.md` G2/T11).
    /// `Access::Server` + `Admin`, like the other replication verbs: it can
    /// enqueue every repository on the box and it holds an admin thread for
    /// up to the timeout.
    async fn flush_replication(
        &self,
        request: Request<FlushReplicationRequest>,
    ) -> Result<Response<FlushReplicationResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx.require(
            &caller,
            Access::Server,
            GrantLevel::Admin,
            "Mirror.FlushReplication",
        )?;
        let req = request.into_inner();
        match self.ctx.control.replication_flush(&req) {
            // **The twin FOLLOWS the log** (it pulls; no push replicator runs here).
            // MEASURED loki 2026-09-26 (the pro kit's paid growth on the mock):
            // every growth was refused at this flush — "no replicator is running on
            // this box" — on an estate whose twin had applied the primary's head a
            // minute before. The proof there is the twins' own reports: every enrolled
            // twin has applied the head this flush began at, waited for, bounded.
            Err(crate::seam::ControlError::FailedPrecondition(why)) if why.starts_with("no replicator is running") => {
                self.log_twin_flush(&req, why).await
            }
            other => Ok(Response::new(other?)),
        }
    }
}

impl MirrorService {
    /// [`Mirror::flush_replication`] for a LOG twin: capture the head, wait until every
    /// enrolled twin (`TWIN_GROUP`) has reported applying it (their LogHead polls,
    /// `heartbeats`), within `timeout_secs` (0 = 120 s). No twin enrolled: the
    /// replicator's own refusal stands (`no_replicator`).
    async fn log_twin_flush(&self, req: &FlushReplicationRequest, no_replicator: String) -> Result<Response<FlushReplicationResponse>, Status> {
        let members: Vec<String> = self.ctx.control.list_group_members(crate::TWIN_GROUP).map(|ps| ps.into_iter().map(|p| p.id.to_string()).collect()).unwrap_or_default();
        if members.is_empty() {
            return Err(crate::seam::ControlError::FailedPrecondition(no_replicator).into());
        }
        let began = estate_clock::now_unix_ms();
        let started = std::time::Instant::now();
        let head = self.ctx.control.log_head()?.head_seq;
        let ceiling = std::time::Duration::from_secs(if req.timeout_secs == 0 { 120 } else { u64::from(req.timeout_secs) });
        let verdict = loop {
            match crate::heartbeat::log_twins_at(&self.ctx.heartbeats.applied_of(&members), head) {
                Ok(v) => break Ok(v),
                Err(why) if started.elapsed() >= ceiling => break Err(why),
                Err(_) => tokio::time::sleep(std::time::Duration::from_millis(250)).await,
            }
        };
        let caught_up = verdict.is_ok();
        let lowest = self.ctx.heartbeats.applied_of(&members).iter().filter_map(|(_, a)| *a).min().unwrap_or(0);
        Ok(Response::new(FlushReplicationResponse {
            caught_up,
            pending: head.saturating_sub(lowest),
            flushed_at_unix_ms: estate_clock::now_unix_ms(),
            waited_ms: started.elapsed().as_millis() as u64,
            last_success_unix_ms: if caught_up { began } else { 0 },
            verdict: match verdict {
                Ok(v) => format!("LOG twin: {v}"),
                Err(why) => format!("LOG twin NOT caught up at head {head} after {} s: {why}", ceiling.as_secs()),
            },
            ..FlushReplicationResponse::default()
        }))
    }
}

/// **How long one snapshot piece may wait for the twin before the snapshot is abandoned.**
///
/// The producer holds `Log::freeze()` for the whole of `Mirror.SnapshotFetch`, so while a
/// snapshot streams EVERY framed write on the primary is stopped — every push, every ref
/// update, every control-plane mutation. That is accepted for as long as the twin is taking
/// pieces: a 44 GiB snapshot legitimately takes a long time and the freeze is the price of a
/// consistent `at`.
///
/// It is NOT accepted when the twin has stopped taking them. `tx.blocking_send` on a full
/// channel waits for ever, and a twin that panics, loses its cable or black-holes its socket
/// leaves the connection open until TCP gives up — with the primary's writes frozen the whole
/// time, and nothing anywhere saying why. This is the deadline on NO PROGRESS, not on the
/// transfer: a slow twin that keeps taking pieces is never cut off.
pub const SNAPSHOT_STALL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(120);

/// Hand one piece to the stream, or give up by name ([`SNAPSHOT_STALL_TIMEOUT`]). Called on the
/// blocking thread that runs the producer, so the freeze it is holding is released by the error.
fn send_piece(
    rt: &tokio::runtime::Handle,
    tx: &tokio::sync::mpsc::Sender<Result<LogChunk, Status>>,
    chunk: LogChunk,
    stall: std::time::Duration,
) -> Result<(), crate::seam::ControlError> {
    rt.block_on(async {
        match tokio::time::timeout(stall, tx.send(Ok(chunk))).await {
            Ok(Ok(())) => Ok(()),
            Ok(Err(_)) => Err(crate::seam::ControlError::Unavailable("the twin hung up mid-snapshot".into())),
            Err(_) => Err(crate::seam::ControlError::Unavailable(format!(
                "the twin took no snapshot piece for {} s; the snapshot is abandoned so this primary's \
                 writes resume — the twin asks again when it is back",
                stall.as_secs()
            ))),
        }
    })
}

/// **How long one `Mirror.LogFetch` piece may wait for its reader** (twin-streaming,
/// 2026-10-06). Nothing on the primary is frozen while a range streams — unlike a snapshot — so
/// this only bounds how long a reader that stopped reading keeps a blocking thread and an open
/// segment here. The twin itself gives up on a quiet stream after `FETCH_IDLE_WITHIN` and asks
/// again.
pub const LOG_FETCH_STALL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(120);

/// Hand one log piece to the stream, or give up by name ([`LOG_FETCH_STALL_TIMEOUT`]).
fn send_log_piece(
    rt: &tokio::runtime::Handle,
    tx: &tokio::sync::mpsc::Sender<Result<LogChunk, Status>>,
    chunk: LogChunk,
    stall: std::time::Duration,
) -> Result<(), crate::seam::ControlError> {
    let seq = chunk.seq;
    rt.block_on(async {
        match tokio::time::timeout(stall, tx.send(Ok(chunk))).await {
            Ok(Ok(())) => Ok(()),
            Ok(Err(_)) => Err(crate::seam::ControlError::Unavailable(format!("the reader hung up mid-fetch, at seq {seq}"))),
            Err(_) => Err(crate::seam::ControlError::Unavailable(format!(
                "the reader took no Mirror.LogFetch piece for {} s (seq {seq}); the fetch is abandoned — \
                 a twin asks again from its own head",
                stall.as_secs()
            ))),
        }
    })
}

/// **Why an encrypted repository is never a mirror**, in the words the person
/// reads (`Mirror.Add`, `Mirror.Sync`).
pub(crate) fn encrypted_is_never_a_mirror(path: &RepoPath) -> String {
    format!(
        "{}/{} is encrypted, so it cannot be a mirror: a mirror copies another server's \
         plaintext into it, and an encrypted repository holds only what its members sealed \
         (push sealed content instead, or turn encryption off first)",
        path.namespace, path.repo
    )
}

#[cfg(test)]
mod tests {
    use super::*;

    /// **A twin that stops reading must not freeze the primary for ever.**
    ///
    /// `snapshot_fetch`'s sink was `tx.blocking_send`, which waits with no deadline, and the
    /// producer on the other side of it is holding `Log::freeze()` — so a twin that stopped
    /// taking pieces (panic, pulled cable, black-holed socket) stopped every framed write on
    /// the PRIMARY until TCP gave up, silently. The sink now gives up by name and the freeze
    /// goes with it.
    #[test]
    fn a_twin_that_stops_taking_pieces_does_not_hold_the_primarys_freeze_for_ever() {
        let rt = tokio::runtime::Builder::new_multi_thread().worker_threads(1).enable_all().build().unwrap();
        let (tx, _rx) = tokio::sync::mpsc::channel::<Result<LogChunk, Status>>(1);
        let piece = || LogChunk { seq: 0, kind: 1, meta: Vec::new(), blob_len: 0, blob: Vec::new(), last: true, restore_head: 0, restore_bytes: 0, log_id: String::new(), offset: None, entry_crc: None, snapshot_item: None, snapshot_id: String::new() };
        // The window is full and the twin is not draining it.
        send_piece(rt.handle(), &tx, piece(), std::time::Duration::from_millis(50)).expect("the first piece fits in the window");

        let started = std::time::Instant::now();
        let err = send_piece(rt.handle(), &tx, piece(), std::time::Duration::from_millis(50)).unwrap_err();
        assert!(started.elapsed() < std::time::Duration::from_secs(5), "the sink waited {:?} for a twin that is not there", started.elapsed());
        let say = format!("{err:?}");
        assert!(say.contains("abandoned") && say.contains("writes resume"), "the refusal says what it gives back: {say}");
    }

    /// A twin that hung up is still its own sentence, not the stall one.
    #[test]
    fn a_twin_that_hung_up_is_named_as_that_and_not_as_a_stall() {
        let rt = tokio::runtime::Builder::new_multi_thread().worker_threads(1).enable_all().build().unwrap();
        let (tx, rx) = tokio::sync::mpsc::channel::<Result<LogChunk, Status>>(1);
        drop(rx);
        let chunk = LogChunk { seq: 0, kind: 1, meta: Vec::new(), blob_len: 0, blob: Vec::new(), last: true, restore_head: 0, restore_bytes: 0, log_id: String::new(), offset: None, entry_crc: None, snapshot_item: None, snapshot_id: String::new() };
        let err = send_piece(rt.handle(), &tx, chunk, SNAPSHOT_STALL_TIMEOUT).unwrap_err();
        assert!(format!("{err:?}").contains("hung up"), "{err:?}");
    }
}