crates/gunnar-grpc/src/service/mirror.rs
//! `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::{
AbdicateReplicationRequest, AddMirrorRequest, AddMirrorResponse, DemoteReplicationRequest,
FlushReplicationRequest, FlushReplicationResponse, LogChunk, LogFetchRequest, LogHeadRequest,
LogHeadResponse, MirrorCredentialKind, MirrorStatusRequest, MirrorStatusResponse,
RemoveMirrorRequest, RemoveMirrorResponse, RemoveReplicationRequest, ReplicationRole,
ReplicationStatusRequest, ReplicationStatusResponse, RepoPath, ResyncReplicationRequest,
ResyncReplicationResponse, SetMirrorCaRequest, SetMirrorCaResponse, SetMirrorCredentialRequest,
SetMirrorCredentialResponse, SetReplicationRequest, SetRoleRequest, 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);
// A32-14, stays a task: one log stream's producer, bounded by its stall timeout; it ends with its stream.
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();
// A32-14, stays a task: one snapshot stream's producer, cut off by its stall timeout; it ends with its stream.
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 set_ca(
&self,
request: Request<SetMirrorCaRequest>,
) -> Result<Response<SetMirrorCaResponse>, Status> {
let caller = self.ctx.caller(&request)?;
let request = request.into_inner();
let path = repo_path(request.repo, "repo")?;
// The credential's grant: what this mirror trusts is as much the admin's as
// what it dials with (GUN-MIRROR-CA).
self.ctx.require(
&caller,
self.mirror_scope(&path),
GrantLevel::Admin,
"Mirror.SetCa",
)?;
let mirror = self.ctx.control.mirror_set_ca(&path, &request.ca_pem)?;
Ok(Response::new(SetMirrorCaResponse {
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)) {
// A registration whose repository the first sync is still to create has no
// repository to hold a grant on yet: the namespace answers, as for Sync.
(Some(path), _) => self.mirror_scope(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:?}");
}
}