use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
use anyhow::{anyhow, Context};
use gunnar_core::control::BlobId;
use gunnar_core::txlog::{self, Entry, Kind, Log, Meta, RefEdit};
use gunnar_core::RepoRef;
use gunnar_store::git_storage_trait::{RefCas, RefRejection, RefUpdate};
use crate::stores::ServerStores;
pub const DEFAULT_FETCH_BYTES: u64 = 64 << 20;
pub const MAX_FETCH_ENTRIES: u64 = 10_000;
pub const GROUP_ENTRIES: usize = 2048;
pub const GROUP_BYTES: u64 = 32 << 20;
pub const GROUP_AGE: Duration = Duration::from_millis(500);
pub const MARK_EVERY: u64 = 4096;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TwinStatus {
pub applied_seq: u64,
pub primary_head_seq: u64,
pub state: State,
pub why: String,
pub last_answer_unix_ms: i64,
pub last_level_unix_ms: i64,
}
impl TwinStatus {
pub fn caught_up(&self) -> bool {
self.state == State::Following && self.applied_seq == self.primary_head_seq
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum State {
Following,
SnapshotNeeded,
Unreachable,
Stalled,
Diverged,
Maintenance,
Recovering,
Resyncing,
}
impl std::fmt::Display for State {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(match self {
State::Following => "following",
State::SnapshotNeeded => "snapshot-needed",
State::Unreachable => "unreachable",
State::Stalled => "stalled",
State::Diverged => "diverged",
State::Maintenance => "maintenance",
State::Recovering => "recovering",
State::Resyncing => "resyncing",
})
}
}
static STATUS: OnceLock<Mutex<HashMap<PathBuf, TwinStatus>>> = OnceLock::new();
fn status_slots() -> std::sync::MutexGuard<'static, HashMap<PathBuf, TwinStatus>> {
STATUS.get_or_init(|| Mutex::new(HashMap::new())).lock().unwrap_or_else(|p| p.into_inner())
}
pub fn status(root: &Path) -> Option<TwinStatus> {
status_slots().get(root).cloned()
}
pub(crate) fn set_status(root: &Path, mut s: TwinStatus) {
let mut slots = status_slots();
let before = slots.get(root).map(|p| p.last_level_unix_ms).unwrap_or(0);
s.last_level_unix_ms = if s.state == State::Following && s.last_answer_unix_ms > 0 && s.applied_seq >= s.primary_head_seq {
s.last_answer_unix_ms
} else {
before
};
slots.insert(root.to_path_buf(), s);
}
pub fn fetch_range(log: &Log, from: u64, to: u64, max_bytes: u64) -> txlog::Result<Vec<Entry>> {
let head = log.head_seq();
let to = if to == 0 { head } else { to.min(head) };
let cap = if max_bytes == 0 { DEFAULT_FETCH_BYTES } else { max_bytes };
let mut out = Vec::new();
let mut bytes = 0u64;
let mut seq = from;
while seq <= to {
let e = log.read(seq)?;
bytes += (e.blob.len() + e.meta.len()) as u64;
out.push(e);
seq += 1;
if bytes >= cap || out.len() as u64 >= MAX_FETCH_ENTRIES {
break;
}
}
if out.is_empty() && from > head {
return Err(txlog::Error::BeyondHead { seq: from, head });
}
Ok(out)
}
pub const PIECE_BYTES: usize = 1 << 20;
pub fn check_range(log: &Log, from: u64, to: u64) -> txlog::Result<()> {
let head = log.head_seq();
if from > head {
return Err(txlog::Error::BeyondHead { seq: from, head });
}
let to = if to == 0 { head } else { to.min(head) };
let first = log.first_seq();
if from <= to && from < first {
return Err(txlog::Error::NotRetained { seq: from, first });
}
Ok(())
}
pub fn stream_range(log: &Log, from: u64, to: u64, max_bytes: u64, into: &mut dyn FrameSink) -> anyhow::Result<u64> {
stream_range_at(log, from, 0, to, max_bytes, into)
}
pub fn stream_range_at(log: &Log, from: u64, from_offset: u64, to: u64, max_bytes: u64, into: &mut dyn FrameSink) -> anyhow::Result<u64> {
check_range(log, from, to)?;
let head = log.head_seq();
let to = if to == 0 { head } else { to.min(head) };
let cap = if max_bytes == 0 { DEFAULT_FETCH_BYTES } else { max_bytes };
let (mut bytes, mut sent, mut seq) = (0u64, 0u64, from);
while seq <= to {
let mut begun = false;
let start = if seq == from && from_offset > 0 {
let l = log.locate_blob(seq)?;
if l.kind == Kind::Redacted as u16 || from_offset > l.blob_len { 0 } else { from_offset }
} else {
0
};
let at = log.read_pieces_from::<anyhow::Error>(seq, start, PIECE_BYTES, |l, piece, last| {
if !begun {
into.begin_at(l.seq, l.kind, &l.meta, l.blob_len, start)?;
begun = true;
}
if last {
if let Some(crc) = l.payload_crc {
into.checked(crc)?;
}
}
if !piece.is_empty() {
into.piece(piece)?;
}
if last {
into.end()?;
}
Ok(())
})?;
bytes += at.blob_len + at.meta.len() as u64;
sent += 1;
seq += 1;
if bytes >= cap || sent >= MAX_FETCH_ENTRIES {
break;
}
}
Ok(sent)
}
pub trait FrameSink {
fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()>;
fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()>;
fn end(&mut self) -> anyhow::Result<()>;
fn begin_at(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64, offset: u64) -> anyhow::Result<()> {
anyhow::ensure!(offset == 0, "seq {seq}: this sink takes whole entries, not one from byte {offset}");
self.begin(seq, kind, meta, blob_len)
}
fn checked(&mut self, _payload_crc: u32) -> anyhow::Result<()> {
Ok(())
}
}
#[derive(Debug)]
pub struct Gather {
pub out: Vec<Entry>,
open: Option<Entry>,
declared: u64,
held: u64,
max_entry: u64,
max_total: u64,
}
impl Default for Gather {
fn default() -> Self {
Gather::bounded(GATHER_MAX_ENTRY, GATHER_MAX_TOTAL)
}
}
pub const GATHER_MAX_ENTRY: u64 = 256 << 20;
pub const GATHER_MAX_TOTAL: u64 = 512 << 20;
impl Gather {
pub fn bounded(max_entry: u64, max_total: u64) -> Gather {
Gather { out: Vec::new(), open: None, declared: 0, held: 0, max_entry, max_total }
}
}
impl FrameSink for Gather {
fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()> {
anyhow::ensure!(self.open.is_none(), "seq {seq} began inside another entry");
anyhow::ensure!(
blob_len <= self.max_entry,
"seq {seq} declares {blob_len} bytes, more than the {} one gathered entry may hold; it is streamed, not gathered",
self.max_entry
);
anyhow::ensure!(
self.held.saturating_add(blob_len) <= self.max_total,
"seq {seq} ({blob_len} bytes) would take this gather past the {} bytes it may hold in all ({} held)",
self.max_total,
self.held
);
self.declared = blob_len;
self.open = Some(Entry { seq, kind, meta: meta.to_vec(), blob: Vec::with_capacity(gunnar_proto::reserve_for(blob_len)) });
Ok(())
}
fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()> {
let open = self.open.as_mut().ok_or_else(|| anyhow!("a piece arrived outside an entry"))?;
let after = open.blob.len() as u64 + bytes.len() as u64;
anyhow::ensure!(after <= self.declared, "seq {}: {after} bytes arrived for an entry that declared {}", open.seq, self.declared);
open.blob.extend_from_slice(bytes);
self.held += bytes.len() as u64;
Ok(())
}
fn end(&mut self) -> anyhow::Result<()> {
let e = self.open.take().ok_or_else(|| anyhow!("an entry ended that never began"))?;
self.out.push(e);
Ok(())
}
}
#[derive(Debug, Default)]
pub struct ChunkFeed {
open: Option<(u64, u16, u64, u64)>,
}
impl ChunkFeed {
pub fn take(&mut self, chunk: &gunnar_proto::LogChunk, into: &mut dyn FrameSink) -> anyhow::Result<()> {
let kind = u16::try_from(chunk.kind).map_err(|_| anyhow!("seq {}: kind {} is not a frame kind", chunk.seq, chunk.kind))?;
match self.open {
Some((seq, k, len, _)) if seq == chunk.seq => anyhow::ensure!(
k == kind && len == chunk.blob_len,
"seq {seq}: a later piece says kind {kind} of {} bytes, the first said kind {k} of {len}",
chunk.blob_len
),
Some((seq, ..)) => anyhow::bail!("the primary moved to seq {} with seq {seq} unfinished", chunk.seq),
None => {
into.begin(chunk.seq, kind, &chunk.meta, chunk.blob_len)?;
self.open = Some((chunk.seq, kind, chunk.blob_len, 0));
}
}
let Some((seq, _, len, got)) = self.open.as_mut() else { unreachable!("an entry is open") };
*got += chunk.blob.len() as u64;
anyhow::ensure!(*got <= *len, "seq {seq}: more than the {len} blob bytes it declared arrived");
if !chunk.blob.is_empty() {
into.piece(&chunk.blob)?;
}
if chunk.last {
anyhow::ensure!(*got == *len, "seq {seq}: {got} of {len} blob bytes arrived");
self.open = None;
into.end()?;
}
Ok(())
}
pub fn finish(&self) -> anyhow::Result<()> {
match self.open {
Some((seq, _, len, got)) => Err(anyhow!("the stream ended inside seq {seq}, {got} of {len} blob bytes in")),
None => Ok(()),
}
}
}
pub async fn feed_log_stream<S, E>(
stream: &mut S,
idle: Option<(Duration, &(dyn Fn() -> anyhow::Error + Sync))>,
said: &(dyn Fn(E) -> anyhow::Error + Sync),
into: &mut dyn FrameSink,
) -> anyhow::Result<()>
where
S: tokio_stream::Stream<Item = Result<gunnar_proto::LogChunk, E>> + Unpin,
{
use tokio_stream::StreamExt as _;
let mut feed = ChunkFeed::default();
loop {
let next = match idle {
Some((within, quiet)) => tokio::time::timeout(within, stream.next()).await.map_err(|_| quiet())?,
None => stream.next().await,
};
match next.transpose().map_err(said)? {
Some(chunk) => feed.take(&chunk, into)?,
None => return feed.finish(),
}
}
}
pub trait LogSource: Send + Sync {
fn head(&self) -> anyhow::Result<(u64, u64)>;
fn alternatives(&self) -> bool {
false
}
fn head_and_identity(&self) -> anyhow::Result<(u64, u64, Option<String>)> {
let (first, head) = self.head()?;
Ok((first, head, None))
}
fn head_identity_generation(&self) -> anyhow::Result<(u64, u64, Option<String>, u64)> {
let (first, head, id) = self.head_and_identity()?;
Ok((first, head, id, 0))
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>>;
fn fetch_into(&self, from: u64, to: u64, max_bytes: u64, into: &mut dyn FrameSink) -> anyhow::Result<()> {
for e in self.fetch(from, to, max_bytes)? {
into.begin(e.seq, e.kind, &e.meta, e.blob.len() as u64)?;
if !e.blob.is_empty() {
into.piece(&e.blob)?;
}
into.end()?;
}
Ok(())
}
fn snapshot(&self, _into: &mut snapshot::Stager) -> anyhow::Result<bool> {
Ok(false)
}
}
impl LogSource for Arc<Log> {
fn head(&self) -> anyhow::Result<(u64, u64)> {
Ok((self.first_seq(), self.head_seq()))
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
Ok(fetch_range(self, from, to, max_bytes)?)
}
fn fetch_into(&self, from: u64, to: u64, max_bytes: u64, into: &mut dyn FrameSink) -> anyhow::Result<()> {
stream_range(self, from, to, max_bytes, into).map(drop)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Applied {
Done,
Lost(String),
Unknown(u16),
Stored,
}
pub fn twin_applies(kind: u16) -> bool {
matches!(Kind::from_u16(kind), Some(Kind::Kek | Kind::ControlBlob | Kind::ControlDoc | Kind::Secret | Kind::Erase | Kind::Activity))
}
pub fn take(stores: &ServerStores, e: &Entry) -> anyhow::Result<Applied> {
take_raw(stores, e.seq, e.kind, &e.meta, &e.blob)
}
pub fn take_raw(stores: &ServerStores, seq: u64, kind: u16, meta: &[u8], blob: &[u8]) -> anyhow::Result<Applied> {
if stores.is_twin() && Kind::from_u16(kind).is_some() && !twin_applies(kind) {
Meta::from_bytes(meta).with_context(|| format!("seq {seq}: the frame's meta does not parse"))?;
return Ok(Applied::Stored);
}
apply_raw(stores, seq, kind, meta, blob)
}
pub fn take_logged(stores: &ServerStores, seq: u64, kind: u16, meta: &[u8]) -> anyhow::Result<Applied> {
if stores.is_twin() && !twin_applies(kind) {
return take_raw(stores, seq, kind, meta, &[]);
}
let e = stores.txlog().read(seq)?;
take(stores, &e)
}
fn oid_bytes(hex_text: &str) -> anyhow::Result<Vec<u8>> {
hex::decode(hex_text).map_err(|e| anyhow!("oid {hex_text}: {e}"))
}
fn repo_of(stores: &ServerStores, key: &str) -> anyhow::Result<gunnar_core::Repo> {
let (ns, name) = key
.split_once('/')
.ok_or_else(|| anyhow!("repository key {key:?} is not namespace/name"))?;
let path = RepoRef::new(ns, name).with_context(|| format!("repository key {key:?}"))?;
stores
.catalog()
.get(&path)?
.ok_or_else(|| anyhow!("the frame names {key}, which the catalog does not hold yet — the catalog row must precede the first pack"))
}
pub fn apply(stores: &ServerStores, e: &Entry) -> anyhow::Result<Applied> {
apply_raw(stores, e.seq, e.kind, &e.meta, &e.blob)
}
pub fn apply_raw(stores: &ServerStores, seq: u64, kind: u16, meta: &[u8], blob: &[u8]) -> anyhow::Result<Applied> {
struct E<'a> {
seq: u64,
blob: &'a [u8],
}
let e = E { seq, blob };
let Some(kind) = Kind::from_u16(kind) else {
if is_restoring(stores.root()) {
anyhow::bail!(
"seq {seq}: frame kind {kind} is not one this build can apply — the log that is being \
restored was written by a NEWER gunnar. The frame is durable at this box's head and \
nothing is lost; restore again from a build that knows kind {kind}. This box stays \
restoring and originates nothing."
);
}
return Ok(Applied::Unknown(kind));
};
if kind == Kind::Redacted {
return Ok(Applied::Done);
}
let meta = Meta::from_bytes(meta).with_context(|| format!("seq {seq}"))?;
match (kind, meta) {
(Kind::Erase, Meta::Erase { repos, .. }) => {
stores.txlog().erase(seq, &repos).with_context(|| format!("seq {seq}: erasing {} repositor(ies) from the log", repos.len()))?;
Ok(Applied::Done)
}
(Kind::Pack, Meta::Pack { repo, .. }) => {
let r = repo_of(stores, &repo)?;
let store = stores.repo_store(&r)?;
store.unlogged().put_pack(e.blob).with_context(|| format!("seq {}: pack into {repo}", e.seq))?;
Ok(Applied::Done)
}
(Kind::Refs, Meta::Refs { repo, atomic, edits }) => {
let r = repo_of(stores, &repo)?;
let store = stores.repo_store(&r)?;
let git = store.unlogged();
if atomic {
let olds: Vec<Option<Vec<u8>>> = edits
.iter()
.map(|ed| ed.old.clone().flatten().map(|h| oid_bytes(&h)).transpose())
.collect::<anyhow::Result<_>>()?;
let news: Vec<Option<Vec<u8>>> = edits
.iter()
.map(|ed| ed.new.as_deref().map(oid_bytes).transpose())
.collect::<anyhow::Result<_>>()?;
let peeled: Vec<Option<Vec<u8>>> = edits
.iter()
.map(|ed| ed.peeled.as_deref().map(oid_bytes).transpose())
.collect::<anyhow::Result<_>>()?;
let cas: Vec<RefCas<'_>> = edits
.iter()
.enumerate()
.map(|(i, ed)| RefCas {
name: ed.name.clone(),
old: olds[i].as_deref(),
new: news[i].as_deref(),
peeled: peeled[i].clone(),
})
.collect();
match git.put_refs_cas(&cas) {
Ok(_) => Ok(Applied::Done),
Err(err) => match err.downcast_ref::<RefRejection>() {
Some(rej) => Ok(Applied::Lost(format!("{rej:?}"))),
None => Err(err).with_context(|| format!("seq {}: refs into {repo}", e.seq)),
},
}
} else {
let mut plain: Vec<RefUpdate> = Vec::new();
for ed in &edits {
if ed.name == "HEAD" {
if let Some(target) = &ed.symref {
git.set_head(target).with_context(|| format!("seq {}: HEAD of {repo}", e.seq))?;
continue;
}
}
plain.push(update_of_edit(ed));
}
if !plain.is_empty() {
git.put_refs(&plain).with_context(|| format!("seq {}: refs into {repo}", e.seq))?;
}
Ok(Applied::Done)
}
}
(Kind::ControlBlob, Meta::ControlBlob { id }) => {
let got = stores.raw_control_store().put_blob(e.blob)?;
let want = BlobId::from_hex(&id).ok_or_else(|| anyhow!("seq {}: blob id {id:?} is not hex", e.seq))?;
if got != want {
return Err(anyhow!("seq {}: the blob hashes to {} but the frame says {id}", e.seq, hex::encode(got.as_bytes())));
}
Ok(Applied::Done)
}
(Kind::ControlDoc, Meta::ControlDoc { name, expected, new }) => {
let mut expected = expected
.as_deref()
.map(|h| BlobId::from_hex(h).ok_or_else(|| anyhow!("seq {}: expected {h:?} is not hex", e.seq)))
.transpose()?;
let new = BlobId::from_hex(&new).ok_or_else(|| anyhow!("seq {}: new {new:?} is not hex", e.seq))?;
if expected.is_none() && fresh_take(stores.root(), &name)? {
expected = stores.raw_control_store().read_doc(&name)?;
}
if stores.raw_control_store().commit_doc(&name, expected, new)? {
Ok(Applied::Done)
} else {
Ok(Applied::Lost(format!("document {name}")))
}
}
(Kind::Secret, Meta::Secret { key, deleted }) => {
use gunnar_store::SecretStore as _;
if deleted {
stores.raw_secrets().delete(&key)?;
} else {
stores.raw_secrets().put(&key, e.blob)?;
}
Ok(Applied::Done)
}
(Kind::LfsPut, Meta::Lfs { repo, oid, size }) => {
use gunnar_lfs::LfsStore as _;
use std::io::Write as _;
let r = repo_of(stores, &repo)?;
let store = stores.repo_store(&r)?;
let lfs = gunnar_lfs::FsLfsStore::open_or_init(store.git_dir())?;
let id = gunnar_lfs::Oid::parse(&oid).map_err(|err| anyhow!("seq {}: {oid}: {err}", e.seq))?;
if lfs.size_of(&id)?.is_some() {
return Ok(Applied::Done);
}
let mut w = lfs.begin_write(&id, size)?;
w.write_all(e.blob)?;
w.commit()?;
Ok(Applied::Done)
}
(Kind::Activity, Meta::Activity {}) => {
let rows = gunnar_core::activity::decode_batch(e.blob).with_context(|| format!("seq {}: last-seen stamps", e.seq))?;
stores.activity().absorb(rows).with_context(|| format!("seq {}: storing last-seen stamps", e.seq))?;
Ok(Applied::Done)
}
(Kind::Kek, Meta::Kek {}) => {
write_kek(stores.control_dir(), e.blob).with_context(|| format!("seq {}: the KEK", e.seq))?;
stores.adopt_kek(e.blob).with_context(|| format!("seq {}: holding the KEK", e.seq))?;
Ok(Applied::Done)
}
(Kind::Append, Meta::Append { log }) => {
let path = under_root(stores.root(), &log)?;
append_line_once(&path, e.blob).with_context(|| format!("seq {}: a line of {log}", e.seq))?;
Ok(Applied::Done)
}
(Kind::File, Meta::File { path, deleted }) => {
let full = under_root(stores.root(), &path)?;
if deleted {
match std::fs::remove_file(&full) {
Ok(()) => {}
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
Err(err) => return Err(err).with_context(|| format!("seq {}: removing {path}", e.seq)),
}
} else {
let dir = full.parent().expect("under the root");
std::fs::create_dir_all(dir)?;
let name = full.file_name().expect("a file").to_string_lossy().into_owned();
crate::durable::write_atomic_unframed(dir, &name, e.blob).with_context(|| format!("seq {}: {path}", e.seq))?;
}
Ok(Applied::Done)
}
(Kind::LfsDelete, Meta::Lfs { repo, oid, .. }) => {
use gunnar_lfs::LfsStore as _;
let (ns, name) = repo.split_once('/').ok_or_else(|| anyhow!("seq {}: repository key {repo:?} is not namespace/name", e.seq))?;
let dir = stores.repo_dir(&RepoRef::new(ns, name)?)?;
if !dir.exists() {
return Ok(Applied::Done);
}
let id = gunnar_lfs::Oid::parse(&oid).map_err(|err| anyhow!("seq {}: {oid}: {err}", e.seq))?;
gunnar_lfs::FsLfsStore::open_or_init(&dir)?.delete(&id).with_context(|| format!("seq {}: deleting LFS {oid} of {repo}", e.seq))?;
Ok(Applied::Done)
}
(Kind::RepoStore, Meta::RepoStore { repo, op, .. }) => {
let (ns, name) = repo.split_once('/').ok_or_else(|| anyhow!("seq {}: repository key {repo:?} is not namespace/name", e.seq))?;
let path = RepoRef::new(ns, name)?;
match op.as_str() {
"purged" => {
stores.reclaim_repo(&path).with_context(|| format!("seq {}: purging the store of {repo}", e.seq))?;
}
"created" => {
if let Some(r) = stores.catalog().get(&path)? {
stores.repo_store(&r)?;
}
}
other => return Err(anyhow!("seq {}: RepoStore op {other:?} is neither created nor purged", e.seq)),
}
Ok(Applied::Done)
}
(kind, meta) => Err(anyhow!("seq {}: kind {kind:?} with meta {meta:?} do not belong together", e.seq)),
}
}
fn update_of_edit(ed: &RefEdit) -> RefUpdate {
RefUpdate {
name: ed.name.clone(),
target: ed.new.clone(),
peeled: ed.peeled.clone(),
symref_target: ed.symref.clone(),
}
}
pub const TWIN_CONFIG: &str = "twin.json";
pub const TWIN_KEY: &str = "twin.key";
pub const TWIN_GROUP: &str = "twin";
pub const TWIN_OWED_FILE: &str = "twin-owed.json";
pub const FRESH_DOCS: &str = "txlog-fresh-docs.json";
pub const RESTORING: &str = "txlog-restoring";
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct TwinConfig {
pub primary: String,
pub joined_at_unix_ms: i64,
}
impl TwinConfig {
pub fn load(root: &Path) -> anyhow::Result<Option<Self>> {
match std::fs::read(root.join(TWIN_CONFIG)) {
Ok(bytes) => Ok(Some(serde_json::from_slice(&bytes).with_context(|| format!("{} does not parse", root.join(TWIN_CONFIG).display()))?)),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(e) => Err(e).with_context(|| format!("reading {}", root.join(TWIN_CONFIG).display())),
}
}
pub fn save(&self, root: &Path) -> anyhow::Result<()> {
crate::durable::write_atomic_unframed(root, TWIN_CONFIG, &serde_json::to_vec_pretty(self)?)
}
}
pub fn is_twin_root(root: &Path) -> bool {
root.join(TWIN_CONFIG).exists()
}
pub const RECHARGE_REL: &str = "pairing/recharge_from";
pub fn is_restoring(root: &Path) -> bool {
root.join(RESTORING).exists()
}
pub const LOG_ID: &str = "txlog-id";
pub const LOG_GENERATION: &str = "txlog-generation";
pub const PINNED_GENERATION: &str = "twin-primary-generation";
fn read_generation(root: &Path, name: &str) -> anyhow::Result<u64> {
let path = root.join(name);
match std::fs::read_to_string(&path) {
Ok(raw) => raw
.trim()
.parse::<u64>()
.map_err(|e| anyhow::anyhow!("{} holds {:?}, which is not a generation: {e}", path.display(), raw.trim())),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(0),
Err(e) => Err(anyhow::Error::from(e).context(format!("reading {}", path.display()))),
}
}
fn forget_generation(root: &Path, name: &str) -> anyhow::Result<()> {
match std::fs::remove_file(root.join(name)) {
Ok(()) => crate::durable::sync_dir(root),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(anyhow::Error::from(e).context(format!("removing {}", root.join(name).display()))),
}
}
pub fn log_generation(root: &Path) -> anyhow::Result<u64> {
read_generation(root, LOG_GENERATION)
}
pub fn pinned_generation(root: &Path) -> anyhow::Result<u64> {
read_generation(root, PINNED_GENERATION)
}
fn stamp_generation(root: &Path) -> anyhow::Result<u64> {
let before = log_generation(root)?;
let generation = u64::try_from(now_ms()).unwrap_or(0).max(before.saturating_add(1));
crate::durable::write_atomic_unframed(root, LOG_GENERATION, format!("{generation}\n").as_bytes())?;
Ok(generation)
}
fn pin_generation(root: &Path, generation: u64) -> anyhow::Result<()> {
if generation <= pinned_generation(root)? {
return Ok(());
}
crate::durable::write_atomic_unframed(root, PINNED_GENERATION, format!("{generation}\n").as_bytes())
}
pub const LOG_FENCED: &str = "txlog-fenced";
pub const GENERATION_PROOF: &str = "twin-generation-proof";
pub fn is_fenced(root: &Path) -> bool {
match std::fs::symlink_metadata(root.join(LOG_FENCED)) {
Ok(_) => true,
Err(e) => e.kind() != std::io::ErrorKind::NotFound,
}
}
pub fn fenced_why(root: &Path) -> Option<String> {
is_fenced(root).then(|| {
let said = std::fs::read_to_string(root.join(LOG_FENCED)).unwrap_or_default();
let said = said.trim();
if said.is_empty() {
format!("this box is fenced off as an OLD generation of its log ({LOG_FENCED} is on its data root)")
} else {
format!("{said} ({LOG_FENCED} is on its data root)")
}
})
}
pub fn generation_proof(root: &Path) -> anyhow::Result<Option<gunnar_proto::GenerationProof>> {
let path = root.join(GENERATION_PROOF);
let raw = match std::fs::read_to_string(&path) {
Ok(raw) => raw,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(e) => return Err(anyhow::Error::from(e).context(format!("reading {}", path.display()))),
};
let lines: Vec<&str> = raw.lines().collect();
let [answer_key, answer_sig, nonce, log_id, first_seq, head_seq, generation] = lines[..] else {
anyhow::bail!("{} holds {} lines, not the 7 of one signed answer", path.display(), lines.len());
};
let number = |what: &str, v: &str| v.trim().parse::<u64>().map_err(|e| anyhow::anyhow!("{}: {what} {v:?} is not a number: {e}", path.display()));
Ok(Some(gunnar_proto::GenerationProof {
answer_key: answer_key.to_owned(),
answer_sig: answer_sig.to_owned(),
nonce: nonce.to_owned(),
log_id: log_id.to_owned(),
first_seq: number("first_seq", first_seq)?,
head_seq: number("head_seq", head_seq)?,
generation: number("generation", generation)?,
}))
}
pub fn remember_generation_proof(root: &Path, nonce: &str, head: &gunnar_proto::LogHeadResponse) -> anyhow::Result<()> {
if head.generation == 0 || head.answer_sig.trim().is_empty() || head.answer_key.trim().is_empty() {
return Ok(());
}
if generation_proof(root).ok().flatten().is_some_and(|kept| kept.generation >= head.generation) {
return Ok(());
}
let one_line = |v: &str| !v.contains('\n') && !v.contains('\r');
anyhow::ensure!(
[head.answer_key.as_str(), head.answer_sig.as_str(), nonce, head.log_id.as_str()].into_iter().all(one_line),
"a signed answer whose key, signature, nonce or log id spans lines is not kept as a proof"
);
let text = format!(
"{}\n{}\n{nonce}\n{}\n{}\n{}\n{}\n",
head.answer_key.trim(),
head.answer_sig.trim(),
head.log_id,
head.first_seq,
head.head_seq,
head.generation
);
crate::durable::write_atomic_unframed(root, GENERATION_PROOF, text.as_bytes())
}
pub fn fence_by_proof(stores: &ServerStores, proof: &gunnar_proto::GenerationProof) -> anyhow::Result<Option<u64>> {
let root = stores.root();
if is_twin_root(root) || stores.is_twin() || is_restoring(root) || is_fenced(root) {
return Ok(None);
}
let own = log_generation(root)?;
if proof.generation <= own {
return Ok(None);
}
let Some(id) = log_id(root) else { return Ok(None) };
if proof.log_id.trim() != id {
tracing::debug!(target: gunnar_core::observe::TARGET_TWIN, theirs = %proof.log_id, mine = %id, "a twin relayed a generation of another log; not this box's business");
return Ok(None);
}
let Some(key) = answering_key(stores) else { return Ok(None) };
let Ok(line) = key.public_key().to_openssh() else { return Ok(None) };
if proof.answer_key.trim() != line.trim()
|| !gunnar_grpc::auth::verify_twin_answer_at(&line, &proof.answer_sig, &proof.nonce, &proof.log_id, proof.first_seq, proof.head_seq, proof.generation)
{
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, generation = proof.generation, "a twin relayed a generation of this log that this box's host key did not sign: ignored");
return Ok(None);
}
let why = format!(
"this box is generation {own} of log {id}; generation {} of the same log exists, restored at {} (unix ms) and signed by this \
box's own host key. This is the OLD main come back after its replacement was restored: it originates nothing and serves no \
log from here, because its writes since that restore belong to no history. Retire it, or re-join it as a twin.",
proof.generation, proof.generation
);
stores.txlog().set_follower(true);
crate::durable::write_atomic_unframed(root, LOG_FENCED, format!("{why}\n").as_bytes()).context("writing the fence")?;
tracing::error!(target: gunnar_core::observe::TARGET_TWIN, event = "twin.fenced_by_generation", own, newer = proof.generation, "{why}");
Ok(Some(proof.generation))
}
fn forget_fence(root: &Path) -> anyhow::Result<()> {
for name in [LOG_FENCED, GENERATION_PROOF] {
match std::fs::remove_file(root.join(name)) {
Ok(()) => crate::durable::sync_dir(root)?,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => return Err(anyhow::Error::from(e).context(format!("removing {}", root.join(name).display()))),
}
}
Ok(())
}
fn answering_key(stores: &ServerStores) -> Option<gunnar_core::ssh_key::PrivateKey> {
static KEY: OnceLock<Option<gunnar_core::ssh_key::PrivateKey>> = OnceLock::new();
KEY.get_or_init(|| {
let secrets = if stores.is_twin() { stores.secrets_unframed() } else { stores.secrets() };
match secrets.and_then(|s| gunnar_core::ensure_host_key(&s).map_err(anyhow::Error::from)) {
Ok(host) => Some(host.private_key().clone()),
Err(err) => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "this box cannot read its host key, so it cannot sign its answers to a twin; a twin that has pinned a key will refuse them");
None
}
}
})
.clone()
}
pub fn sign_head_answer(
stores: &ServerStores,
nonce: &str,
log_id: &str,
first_seq: u64,
head_seq: u64,
generation: u64,
) -> (String, String) {
if nonce.trim().is_empty() {
return (String::new(), String::new());
}
let Some(key) = answering_key(stores) else {
return (String::new(), String::new());
};
let line = match key.public_key().to_openssh() {
Ok(line) => line,
Err(err) => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "this box's host key does not re-encode");
return (String::new(), String::new());
}
};
match gunnar_grpc::auth::sign_twin_answer_at(&key, nonce, log_id, first_seq, head_seq, generation) {
Ok(sig) => (line, sig),
Err(err) => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "signing this answer to a twin");
(String::new(), String::new())
}
}
}
pub fn log_id(root: &Path) -> Option<String> {
let raw = std::fs::read_to_string(root.join(LOG_ID)).ok()?;
let id = raw.trim().to_owned();
if id.is_empty() {
None
} else {
Some(id)
}
}
pub fn mint_log_id(root: &Path) -> anyhow::Result<String> {
if let Some(id) = log_id(root) {
return Ok(id);
}
use gunnar_core::ssh_key::rand_core::Rng as _;
let mut bytes = [0u8; 16];
gunnar_core::os_rng().fill_bytes(&mut bytes);
let id: String = bytes.iter().map(|b| format!("{b:02x}")).collect();
crate::durable::write_atomic_unframed(root, LOG_ID, format!("{id}\n").as_bytes())?;
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, log_id = %id, "this box minted its log's identity: from here a twin can tell this history from any other");
Ok(id)
}
pub fn pin_log_id(root: &Path, id: &str) -> anyhow::Result<()> {
let id = id.trim();
anyhow::ensure!(!id.is_empty(), "a log identity cannot be empty");
if let Some(mine) = log_id(root) {
anyhow::ensure!(
mine == id,
"this data root already holds log {mine} and was asked to hold log {id}; two histories do not become \
one by overwriting a file"
);
return Ok(());
}
crate::durable::write_atomic_unframed(root, LOG_ID, format!("{id}\n").as_bytes())
}
pub const PRIMARY_KEY: &str = "twin-primary.pub";
pub fn pinned_primary_key(root: &Path) -> Option<String> {
let raw = std::fs::read_to_string(root.join(PRIMARY_KEY)).ok()?;
let line = raw.trim().to_owned();
if line.is_empty() {
None
} else {
Some(line)
}
}
pub fn pin_primary_key(root: &Path, line: &str) -> anyhow::Result<()> {
let line = line.trim();
anyhow::ensure!(!line.is_empty(), "a primary's key cannot be empty");
if let Some(mine) = pinned_primary_key(root) {
anyhow::ensure!(mine == line, "this twin already follows a box that answers with {mine}; it was offered {line}");
return Ok(());
}
crate::durable::write_atomic_unframed(root, PRIMARY_KEY, format!("{line}\n").as_bytes())?;
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, key = %line, "this twin has pinned the key its primary answers with; every later answer must be signed by it");
Ok(())
}
fn forget_own_audit_trail(stores: &ServerStores) -> anyhow::Result<()> {
let path = stores.audit_path();
match std::fs::remove_file(&path) {
Ok(()) => {
tracing::info!(
target: gunnar_core::observe::TARGET_TWIN,
path = %path.display(),
"dropped this fresh box's own audit trail: the one being restored starts at its own seq 0"
);
Ok(())
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e).with_context(|| format!("dropping this box's own audit trail {}", path.display())),
}
}
pub fn rematerialise_audit_trail(stores: &ServerStores, just_restored: bool) {
let path = stores.audit_path();
let when = if just_restored { "the first start after a restore" } else { "this start" };
let t0 = std::time::Instant::now();
let outcome = rebuild_audit_trail(stores.txlog(), &path);
let micros = t0.elapsed().as_micros() as u64;
match outcome {
Ok(Rebuilt::Nothing) => {}
Ok(Rebuilt::Intact(n)) => tracing::info!(
target: gunnar_core::observe::TARGET_TWIN,
path = %path.display(),
entries = n,
micros,
"the audit trail is whole at {when}: its tip is the log's newest audit entry"
),
Ok(Rebuilt::Rebuilt { had, now }) => tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
path = %path.display(),
had,
now,
micros,
"the audit trail held {had} of the {now} entries its log carries at {when} — the \
off-stripe copy lost its tail (a hard stop; XFS recovery on the boot disk's P3). \
REBUILT from the log, which did not: the trail is whole again and the next entry \
continues the chain the twin holds"
),
Ok(Rebuilt::Longer { file, log }) => tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
path = %path.display(),
file,
log,
"the audit trail holds MORE entries ({file}) than its log carries ({log}) at {when}; \
left exactly as found — nothing is deleted from an audit trail"
),
Ok(Rebuilt::Diverged { why }) => tracing::error!(
target: gunnar_core::observe::TARGET_TWIN,
path = %path.display(),
why = %why,
"the audit trail and its log DISAGREE at {when}; left exactly as found — nothing is \
guessed into an audit trail. The twin's copy is the one to compare against"
),
Err(err) => tracing::error!(
target: gunnar_core::observe::TARGET_TWIN,
path = %path.display(),
error = %format!("{err:#}"),
"the audit trail could not be checked against its log at {when}; if the off-stripe \
copy lost its tail, the gate appends to a shorter chain than the twin holds"
),
}
}
#[derive(Debug, PartialEq, Eq)]
pub enum Rebuilt {
Nothing,
Intact(usize),
Rebuilt {
had: usize,
now: usize,
},
Longer {
file: usize,
log: usize,
},
Diverged {
why: String,
},
}
enum ViewLine {
Frame(u64),
Snap(usize),
}
pub fn rebuild_audit_trail(log: &Log, path: &Path) -> anyhow::Result<Rebuilt> {
let (append, file) = (Kind::Append as u16, Kind::File as u16);
let audit_meta = Meta::Append { log: txlog::AUDIT_REL.into() }.to_bytes();
let names_trail = |meta: &[u8]| meta.windows(txlog::AUDIT_REL.len()).any(|w| w == txlog::AUDIT_REL.as_bytes());
let mut frames: Vec<u64> = Vec::new();
let mut snapshot: Option<(u64, bool)> = None;
let mut bad_meta: Option<(u64, String)> = None;
log.scan_meta(|seq, kind, meta| {
if kind == append && meta == audit_meta.as_slice() {
frames.push(seq);
return;
}
if (kind != append && kind != file) || !names_trail(meta) {
return;
}
match Meta::from_bytes(meta) {
Ok(Meta::Append { log: rel }) if kind == append && rel == txlog::AUDIT_REL => frames.push(seq),
Ok(Meta::File { path: rel, deleted }) if kind == file && rel == txlog::AUDIT_REL => {
frames.clear();
snapshot = Some((seq, deleted));
}
Ok(_) => {}
Err(e) => bad_meta = bad_meta.take().or(Some((seq, e.to_string()))),
}
})
.context("scanning the log's frame headers")?;
if let Some((seq, e)) = bad_meta {
return Err(anyhow!("frame {seq}'s meta does not parse: {e}"));
}
if frames.is_empty() && snapshot.is_none() {
return Ok(Rebuilt::Nothing);
}
let snap_lines: Vec<Vec<u8>> = match snapshot {
Some((seq, false)) => split_lines(&log.read(seq)?.blob).0,
_ => Vec::new(),
};
let mut view: Vec<ViewLine> = (0..snap_lines.len()).map(ViewLine::Snap).collect();
view.extend(frames.iter().map(|s| ViewLine::Frame(*s)));
let bytes = match std::fs::read(path) {
Ok(b) => b,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Vec::new(),
Err(e) => return Err(e).with_context(|| format!("reading {}", path.display())),
};
let (mut lines, rest) = split_lines(&bytes);
let mut torn = false;
if let Some(rest) = rest {
if serde_json::from_slice::<gunnar_core::AuditEntry>(&rest).is_ok() {
lines.push(rest);
} else {
torn = true;
}
}
let had = lines.len();
let tip: Option<gunnar_core::AuditEntry> = match lines.last() {
Some(l) => Some(serde_json::from_slice(l).with_context(|| format!("the last whole line of {} is not an audit entry", path.display()))?),
None => None,
};
let line_at = |v: &ViewLine| -> anyhow::Result<Vec<u8>> {
Ok(match v {
ViewLine::Frame(seq) => log.read(*seq)?.blob,
ViewLine::Snap(i) => snap_lines[*i].clone(),
})
};
let mut missing: Vec<Vec<u8>> = Vec::new();
let mut met = false;
for (k, v) in view.iter().enumerate().rev() {
let line = line_at(v)?;
let entry: gunnar_core::AuditEntry = serde_json::from_slice(&line).context("an audit line in the log is not an entry")?;
match &tip {
Some(t) if entry.seq == t.seq => {
if entry.hash != t.hash {
return Ok(Rebuilt::Diverged { why: format!("seq {}: the file's hash is {}, the log's {}", t.seq, t.hash, entry.hash) });
}
met = true;
break;
}
Some(t) if entry.seq < t.seq => {
if k + 1 == view.len() {
return Ok(Rebuilt::Longer { file: lines.len(), log: view.len() });
}
return Ok(Rebuilt::Diverged { why: format!("the log skips from seq {} to a newer entry, past the file's tip {}", entry.seq, t.seq) });
}
_ => missing.push(line),
}
}
if missing.is_empty() && !torn {
return Ok(Rebuilt::Intact(had));
}
missing.reverse();
if !met {
let first: gunnar_core::AuditEntry = serde_json::from_slice(&missing[0])?;
let want = tip.as_ref().map_or(0, |t| t.seq + 1);
if first.seq != want {
return Ok(Rebuilt::Diverged { why: format!("the log's audit entries begin at seq {}, the file needs {want} next — a retention hole nothing can bridge", first.seq) });
}
}
let mut out = Vec::with_capacity(bytes.len() + missing.iter().map(|l| l.len() + 1).sum::<usize>());
for l in lines.iter().chain(missing.iter()) {
out.extend_from_slice(l);
out.push(b'\n');
}
let entries: Vec<gunnar_core::AuditEntry> = out
.split(|b| *b == b'\n')
.filter(|l| !l.is_empty())
.map(serde_json::from_slice)
.collect::<Result<_, _>>()
.context("parsing the joined trail")?;
if let Err(e) = gunnar_core::audit::verify_entries(&entries) {
return Ok(Rebuilt::Diverged { why: format!("the file joined to the log's tail does not verify: {e}") });
}
let dir = path.parent().ok_or_else(|| anyhow!("{} has no directory", path.display()))?;
std::fs::create_dir_all(dir).with_context(|| format!("creating {}", dir.display()))?;
let name = path.file_name().ok_or_else(|| anyhow!("{} names no file", path.display()))?.to_string_lossy().into_owned();
crate::durable::write_atomic_unframed(dir, &name, &out).with_context(|| format!("rewriting {}", path.display()))?;
gunnar_core::AuditChain::open(path).map_err(|e| anyhow!("the rebuilt trail does not open: {e}"))?;
Ok(Rebuilt::Rebuilt { had, now: had + missing.len() })
}
fn split_lines(bytes: &[u8]) -> (Vec<Vec<u8>>, Option<Vec<u8>>) {
let (whole, rest) = match bytes.iter().rposition(|b| *b == b'\n') {
Some(i) => (&bytes[..i], &bytes[i + 1..]),
None => (&bytes[..0], bytes),
};
let lines = whole.split(|b| *b == b'\n').filter(|l| !l.iter().all(u8::is_ascii_whitespace)).map(<[u8]>::to_vec).collect();
let rest = (!rest.iter().all(|b| b.is_ascii_whitespace() || *b == 0)).then(|| rest.to_vec());
(lines, rest)
}
fn forget_log_id(root: &Path) -> anyhow::Result<()> {
match std::fs::remove_file(root.join(LOG_ID)) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e).with_context(|| format!("dropping this box's own {LOG_ID}")),
}
}
static GIT_BOUND_AS_TWIN: AtomicBool = AtomicBool::new(false);
pub fn note_git_bind_is_twin_shaped(yes: bool) {
GIT_BOUND_AS_TWIN.store(yes, Ordering::SeqCst);
}
pub fn git_listener_needs_restart(stores: &ServerStores) -> bool {
stores.is_twin() && !GIT_BOUND_AS_TWIN.load(Ordering::SeqCst)
}
pub fn now_ms() -> i64 {
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as i64).unwrap_or(0)
}
pub fn prepare_replay(stores: &ServerStores) -> anyhow::Result<()> {
if !stores.auth().is_empty()? || !stores.auth().bootstrap_window_open()? {
anyhow::bail!("this box has enrolled identities; only a fresh box (bootstrap window open, no key) can take another box's log");
}
if !stores.catalog().list(None)?.is_empty() {
anyhow::bail!("this box holds repositories; only a fresh box can take another box's log");
}
let log = stores.txlog();
let mut names = std::collections::BTreeSet::new();
if log.head_seq() >= log.first_seq() {
for e in log.range(log.first_seq(), log.head_seq())? {
if let Ok(Meta::ControlDoc { name, .. }) = e.meta() {
names.insert(name);
}
}
}
crate::durable::write_atomic_unframed(stores.root(), FRESH_DOCS, &serde_json::to_vec(&names)?)?;
forget_own_secrets(stores)?;
forget_log_id(stores.root())?;
forget_generation(stores.root(), PINNED_GENERATION)?;
forget_fence(stores.root())?;
forget_own_audit_trail(stores)?;
log.set_follower(true);
log.reset(u64::MAX)?;
Ok(())
}
fn forget_own_secrets(stores: &ServerStores) -> anyhow::Result<()> {
use gunnar_store::SecretStore as _;
let rows = stores.control_db().export().context("reading this box's own secrets before it gives up its numbers")?.secrets;
let mut gone = Vec::new();
for (key, value) in &rows {
if value.is_none() {
continue;
}
stores.raw_secrets().delete(key).with_context(|| format!("dropping this box's own secret {key}"))?;
gone.push(key.clone());
}
if !gone.is_empty() {
tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
secrets = ?gone,
"this box gave up the secrets it minted for itself along with its own log numbers: they are sealed under a KEK the replay is about to replace, and what the log carries is the only identity this box keeps"
);
}
Ok(())
}
fn fresh_take(root: &Path, name: &str) -> anyhow::Result<bool> {
let path = root.join(FRESH_DOCS);
let bytes = match std::fs::read(&path) {
Ok(b) => b,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(false),
Err(e) => return Err(e.into()),
};
let mut names: std::collections::BTreeSet<String> = serde_json::from_slice(&bytes)?;
if !names.remove(name) {
return Ok(false);
}
crate::durable::write_atomic_unframed(root, FRESH_DOCS, &serde_json::to_vec(&names)?)?;
Ok(true)
}
fn under_root(root: &Path, rel: &str) -> anyhow::Result<std::path::PathBuf> {
let p = Path::new(rel);
if p.is_absolute() || p.components().any(|c| !matches!(c, std::path::Component::Normal(_))) {
anyhow::bail!("the frame names {rel:?}, which is not a plain path under the data root");
}
if txlog::never_framed(rel) {
anyhow::bail!("the frame names {rel:?}, a cache or per-box path — never applied on a twin");
}
Ok(root.join(p))
}
fn append_line_once(path: &Path, line: &[u8]) -> anyhow::Result<()> {
use std::io::{Read as _, Seek as _, SeekFrom, Write as _};
let mut want = line.to_vec();
want.push(b'\n');
if let Ok(mut f) = std::fs::File::open(path) {
let len = f.metadata()?.len();
if len >= want.len() as u64 {
f.seek(SeekFrom::Start(len - want.len() as u64))?;
let mut tail = vec![0u8; want.len()];
f.read_exact(&mut tail)?;
let boundary = len == want.len() as u64 || {
f.seek(SeekFrom::Start(len - want.len() as u64 - 1))?;
let mut b = [0u8; 1];
f.read_exact(&mut b)?;
b[0] == b'\n'
};
if tail == want && boundary {
return Ok(());
}
}
}
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let mut f = std::fs::OpenOptions::new().create(true).append(true).open(path)?;
f.write_all(&want)?;
f.sync_data()?;
Ok(())
}
fn write_kek(control_dir: &Path, bytes: &[u8]) -> anyhow::Result<()> {
use std::io::Write as _;
anyhow::ensure!(bytes.len() == 32, "a KEK frame carries {} bytes, not 32", bytes.len());
let path = gunnar_core::seal::kek_path(control_dir);
if std::fs::read(&path).ok().as_deref() == Some(bytes) {
return Ok(());
}
std::fs::create_dir_all(control_dir)?;
let tmp = control_dir.join("secrets.kek.tmp");
let mut options = std::fs::OpenOptions::new();
options.write(true).create(true).truncate(true);
{
use std::os::unix::fs::OpenOptionsExt as _;
options.mode(0o600);
}
let mut f = options.open(&tmp)?;
f.write_all(bytes)?;
f.sync_all()?;
drop(f);
std::fs::rename(&tmp, &path)?;
crate::durable::sync_dir(control_dir)
}
pub fn become_twin(stores: &ServerStores, primary: &str) -> anyhow::Result<String> {
let root = stores.root();
if let Some(existing) = TwinConfig::load(root)? {
anyhow::ensure!(existing.primary == primary, "this box is already the twin of {}; a twin is never re-pointed", existing.primary);
return twin_public_key(root);
}
anyhow::ensure!(!primary.trim().is_empty(), "the primary's address is required");
prepare_replay(stores)?;
if !root.join(TWIN_KEY).exists() {
let key = gunnar_core::mint_ed25519()?;
crate::durable::write_atomic_unframed(root, TWIN_KEY, &key)?;
use std::os::unix::fs::PermissionsExt as _;
std::fs::set_permissions(root.join(TWIN_KEY), std::fs::Permissions::from_mode(0o600))?;
}
TwinConfig { primary: primary.trim().to_string(), joined_at_unix_ms: now_ms() }.save(root)?;
stores.mark_twin();
forget_mirrors(stores);
stores.browse().store(true, Ordering::SeqCst);
stores.writes_frozen().store(true, Ordering::SeqCst);
twin_public_key(root)
}
pub fn forget_mirrors(stores: &ServerStores) {
let meta = stores.slow().path_of(crate::slow::META_DIR);
#[allow(unused_mut)]
let mut names = vec![crate::stores::IDENTITY_MIRROR];
#[cfg(feature = "replication")]
names.push(crate::mirror::REPLICATION_CONFIG);
#[cfg(feature = "webhooks")]
names.push(crate::webhook::WEBHOOK_CONFIG);
for name in names {
let path = meta.join(name);
match std::fs::remove_file(&path) {
Ok(()) => tracing::info!(target: gunnar_core::observe::TARGET_TWIN, path = %path.display(), "removed a per-box mirror: a twin keeps none"),
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
Err(err) => tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, path = %path.display(), %err, "could not remove a per-box mirror on a twin"),
}
}
}
pub fn twin_private_key(root: &Path) -> anyhow::Result<gunnar_core::ssh_key::PrivateKey> {
let bytes = std::fs::read(root.join(TWIN_KEY)).with_context(|| format!("reading {}", root.join(TWIN_KEY).display()))?;
gunnar_core::ssh_key::PrivateKey::from_openssh(&bytes).map_err(|e| anyhow!("{TWIN_KEY}: {e}"))
}
pub fn twin_public_key(root: &Path) -> anyhow::Result<String> {
let key = twin_private_key(root)?;
let mut line = key.public_key().to_openssh().map_err(|e| anyhow!("{e}"))?;
line.push_str(" gunnar-twin");
Ok(line)
}
pub const RESTORE_OWNER_LOCAL: &str = "local";
pub const RESTORE_NOT_YOURS: &str = "restore-not-yours";
#[derive(Debug)]
pub struct RestoreNotYours {
pub started_by: String,
pub asked_by: String,
}
impl std::fmt::Display for RestoreNotYours {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"REFUSED [{RESTORE_NOT_YOURS}] this box is mid-restore and the restore was started by {}; {} did not start it \
and cannot resume it. Only the key that started a restore resumes it — run the same replay again with the \
same key (monetize's restore lane, or `gunnar control twin replay -i <that key>`). A box whose starter is \
gone for good is re-imaged, which makes it fresh again; nothing of the stream so far was touched",
self.started_by, self.asked_by
)
}
}
impl std::error::Error for RestoreNotYours {}
fn restoring_marker(owner: &str) -> String {
format!("restoring\nowner {owner}\n")
}
fn restoring_owner(root: &Path) -> anyhow::Result<Option<String>> {
let text = std::fs::read_to_string(root.join(RESTORING)).with_context(|| format!("reading {}", root.join(RESTORING).display()))?;
Ok(text.lines().find_map(|l| l.strip_prefix("owner ")).map(|o| o.trim().to_owned()).filter(|o| !o.is_empty()))
}
fn restore_begin_lock(root: &Path) -> Arc<std::sync::Mutex<()>> {
static LOCKS: std::sync::OnceLock<std::sync::Mutex<std::collections::HashMap<PathBuf, Arc<std::sync::Mutex<()>>>>> = std::sync::OnceLock::new();
let mut locks = LOCKS.get_or_init(Default::default).lock().unwrap_or_else(|p| p.into_inner());
Arc::clone(locks.entry(root.to_path_buf()).or_default())
}
pub fn restore_begin(stores: &ServerStores) -> anyhow::Result<u64> {
restore_begin_by(stores, RESTORE_OWNER_LOCAL)
}
pub fn restore_begin_by(stores: &ServerStores, owner: &str) -> anyhow::Result<u64> {
anyhow::ensure!(
!owner.trim().is_empty() && !owner.contains('\n'),
"a restore needs the name of who is restoring; {owner:?} is not one"
);
let lock = restore_begin_lock(stores.root());
let _held = lock.lock().unwrap_or_else(|p| p.into_inner());
if is_restoring(stores.root()) {
match restoring_owner(stores.root())? {
Some(started_by) if started_by != owner => {
return Err(RestoreNotYours { started_by, asked_by: owner.to_owned() }.into());
}
Some(_) => {}
None => {
tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
owner,
"this box is mid-restore under a marker that names no owner (written before 2026-09-28); {owner} resumes it and owns it from now on"
);
crate::durable::write_atomic_unframed(stores.root(), RESTORING, restoring_marker(owner).as_bytes())?;
}
}
}
snapshot::forget_restore_stager(stores.root());
if is_restoring(stores.root()) {
stores.txlog().set_follower(true);
let head = stores.txlog().head_seq();
if head > 0 && head >= stores.txlog().first_seq() && !snapshot::installing(stores.root()) {
let e = stores.txlog().read(head)?;
apply(stores, &e).with_context(|| format!("re-applying seq {head} before the restore resumes"))?;
}
return Ok(stores.txlog().head_seq() + 1);
}
anyhow::ensure!(!stores.is_twin(), "a twin is never restored into; restore into a fresh primary");
prepare_replay(stores)?;
intake::forget(stores.root())?;
snapshot::forget_restore_snapshot(stores.root())?;
crate::durable::write_atomic_unframed(stores.root(), RESTORING, restoring_marker(owner).as_bytes())?;
Ok(1)
}
pub const RESTORE_BYTES_PER_LOG_BYTE: u64 = 2;
fn reserve_bytes(total: u64) -> u64 {
(total / 20).max(256 << 20)
}
pub fn room_for_restore(root: &Path, announced_bytes: u64) -> anyhow::Result<()> {
if announced_bytes == 0 {
return Ok(());
}
let (total, used) = volume_bytes(root);
if total == 0 {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, root = %root.display(), "the data volume could not be measured; the restore's room is not checked");
return Ok(());
}
let free = total.saturating_sub(used);
let need = announced_bytes.saturating_mul(RESTORE_BYTES_PER_LOG_BYTE).saturating_add(reserve_bytes(total));
if free >= need {
tracing::info!(
target: gunnar_core::observe::TARGET_TWIN,
announced_bytes, need, free, total,
"room for the restore: the data set on this machine holds it"
);
return Ok(());
}
anyhow::bail!(
"NOT ENOUGH ROOM for this restore, and nothing has been written: the sender streams \
{announced} of log, which needs about {need} on this box ({factor}× the log — a restore \
writes every frame into this box's own log AND into the store it builds — plus {reserve} \
the box keeps for its caches and slack), and this data set has {free} free of {total}. \
SHORT BY {short}. This box is being restored onto a SMALLER data set than the one the \
twin's log came from: give it more or larger members and restore again. NOTHING of the \
twin's log has landed — this box's log is still empty and it stays `restoring`, so the \
same replay into a bigger set starts at seq 1 and loses nothing. The estimate is \
DELIBERATELY conservative and a long-lived box's log is larger than the stores it \
builds; if you have read these numbers and this set does hold it, re-run the replay \
with `--ignore-room`.",
announced = human(announced_bytes),
need = human(need),
factor = RESTORE_BYTES_PER_LOG_BYTE,
reserve = human(reserve_bytes(total)),
free = human(free),
total = human(total),
short = human(need - free),
)
}
fn human(n: u64) -> String {
const UNITS: [&str; 5] = ["B", "KiB", "MiB", "GiB", "TiB"];
let mut v = n as f64;
let mut u = 0;
while v >= 1024.0 && u + 1 < UNITS.len() {
v /= 1024.0;
u += 1;
}
if u == 0 {
format!("{n} B")
} else {
format!("{v:.1} {}", UNITS[u])
}
}
pub fn restore_apply(stores: &ServerStores, e: &Entry) -> anyhow::Result<Applied> {
stores.txlog().append_at(e.seq, e.kind, &e.meta, &e.blob)?;
apply(stores, e)
}
pub fn restore_finish(stores: &ServerStores, announced_head: u64, log_id_of_source: &str) -> anyhow::Result<u64> {
finish_restore(stores, announced_head, log_id_of_source, false)
}
pub fn restore_finish_streamed(stores: &ServerStores, announced_head: u64, log_id_of_source: &str) -> anyhow::Result<u64> {
finish_restore(stores, announced_head, log_id_of_source, true)
}
fn finish_restore(stores: &ServerStores, announced_head: u64, log_id_of_source: &str, streamed: bool) -> anyhow::Result<u64> {
let kept = restore_announced(stores.root())?;
let now_id = log_id_of_source.trim();
if streamed {
anyhow::ensure!(
!now_id.is_empty(),
"this restore stream named no log (LogChunk.log_id is empty): a box is restored from one named history, never from whatever numbers fit — this box stays restoring; resend the twin's log with its log id"
);
if let Some(kept) = &kept {
anyhow::ensure!(
!kept.log_id.is_empty(),
"an earlier stream of this restore named no log ({RESTORE_ANNOUNCED} says head {} and no log id): which history this box holds is unknown — this box stays restoring; start the restore again on a fresh box",
kept.head
);
}
}
let Announced { head: kept_head, log_id: kept_id } = kept.unwrap_or_default();
anyhow::ensure!(
kept_id.is_empty() || now_id.is_empty() || kept_id == now_id,
"this restore began as log {kept_id} and a resend is log {now_id}: one restore is one history — this box stays restoring"
);
let announced_head = announced_head.max(kept_head);
let log_id_of_source = if now_id.is_empty() { kept_id.as_str() } else { now_id };
if announced_head > 0 || !log_id_of_source.is_empty() {
crate::durable::write_atomic_unframed(stores.root(), RESTORE_ANNOUNCED, format!("{announced_head} {log_id_of_source}\n").as_bytes())?;
}
anyhow::ensure!(
stores.txlog().head_seq() > 0,
"nothing was restored (the stream carried no entry); this box stays restoring — resend the twin's log"
);
let head = stores.txlog().head_seq();
anyhow::ensure!(
head >= announced_head,
"the stream ended at seq {head} but the sender announced the twin's head {announced_head}; a short stream is not a restore — this box stays restoring, resend the twin's log"
);
let announced_id = log_id_of_source.trim();
if !announced_id.is_empty() {
pin_log_id(stores.root(), announced_id).context("taking the identity of the log this box was restored from")?;
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, log_id = %announced_id, "this box IS log {announced_id} now: the twin it was restored from keeps following it without being re-joined");
}
let generation = stamp_generation(stores.root()).context("stamping this restore's generation of the log")?;
forget_fence(stores.root()).context("a restored box is the newest generation: no fence from an older life")?;
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, generation, "this restore is generation {generation} of its log: a twin refuses any older one from here");
let _ = std::fs::remove_file(stores.root().join(FRESH_DOCS));
crate::durable::write_atomic_unframed(stores.root(), RESEAL, b"reseal\n")?;
crate::durable::remove_durable(stores.root(), RESTORING)?;
let _ = std::fs::remove_file(stores.root().join(RESTORE_ANNOUNCED));
let _ = intake::forget(stores.root());
snapshot::forget_restore_snapshot_or_warn(stores.root());
Ok(stores.txlog().head_seq())
}
pub const RESTORE_ANNOUNCED: &str = "txlog-restore-announced";
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Announced {
pub head: u64,
pub log_id: String,
}
#[derive(Debug, thiserror::Error)]
pub enum RestoreMarkerError {
#[error("the restore marker {path} could not be read ({source}): what this restore was announced to reach is unknown — this box stays restoring")]
Unreadable {
path: std::path::PathBuf,
source: std::io::Error,
},
#[error("the restore marker {path} is corrupt ({what}: {text:?}): what this restore was announced to reach is unknown — this box stays restoring")]
Corrupt {
path: std::path::PathBuf,
what: &'static str,
text: String,
},
}
pub fn restore_announced(root: &Path) -> Result<Option<Announced>, RestoreMarkerError> {
let path = root.join(RESTORE_ANNOUNCED);
let text = match std::fs::read_to_string(&path) {
Ok(t) => t,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(source) => return Err(RestoreMarkerError::Unreadable { path, source }),
};
let corrupt = |what| RestoreMarkerError::Corrupt { path: path.clone(), what, text: text.clone() };
let mut words = text.split_whitespace();
let head = words.next().ok_or_else(|| corrupt("it is empty"))?.parse::<u64>().map_err(|_| corrupt("its head is not a number"))?;
let log_id = words.next().unwrap_or("").to_string();
if words.next().is_some() {
return Err(corrupt("it holds more than a head and a log id"));
}
Ok(Some(Announced { head, log_id }))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AfterRestore {
Reboot,
Operator,
}
pub fn after_restore(is_init: bool, appliance: bool) -> AfterRestore {
if is_init && appliance {
AfterRestore::Reboot
} else {
AfterRestore::Operator
}
}
pub const REBOOT_AFTER_RESTORE: std::time::Duration = std::time::Duration::from_secs(5);
pub fn restart_after_restore() -> AfterRestore {
let how = after_restore(appliance_pid1::net::is_init(), appliance_pid1::net::booted_by_korp_start());
match how {
AfterRestore::Reboot => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, after_s = REBOOT_AFTER_RESTORE.as_secs(), "RESTORED from the twin: this box reboots itself to open as the primary it was");
let spawned = std::thread::Builder::new().name("restore-reboot".into()).spawn(|| {
std::thread::sleep(REBOOT_AFTER_RESTORE);
crate::narrate::reboot_now();
});
if let Err(e) = spawned {
tracing::error!(target: gunnar_core::observe::TARGET_TWIN, error = %e, "the reboot after the restore could not be scheduled; restart this box");
}
}
AfterRestore::Operator => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, "RESTORED from the twin: RESTART this process — it opens as the primary it was (not the appliance's PID 1, so it does not reboot itself)");
}
}
how
}
#[cfg(test)]
mod after_restore_tests {
use super::*;
#[test]
fn only_the_appliances_pid_1_reboots_itself_after_a_restore() {
assert_eq!(after_restore(true, true), AfterRestore::Reboot);
assert_eq!(after_restore(true, false), AfterRestore::Operator, "a container's PID 1 never reboots the host");
assert_eq!(after_restore(false, true), AfterRestore::Operator);
assert_eq!(after_restore(false, false), AfterRestore::Operator);
}
}
pub const RESEAL: &str = "txlog-reseal";
const KEK_NEXT: &str = "secrets.kek.new";
pub fn reseal_pending(root: &Path) -> bool {
root.join(RESEAL).exists()
}
pub fn reseal_after_restore(
root: &Path,
control_dir: &Path,
log: &Log,
rows: &std::collections::BTreeMap<String, Option<Vec<u8>>>,
raw: &dyn gunnar_store::SecretStore,
) -> anyhow::Result<gunnar_core::Kek> {
use gunnar_core::Kek;
let path = gunnar_core::seal::kek_path(control_dir);
let key32 = |bytes: Vec<u8>, what: &Path| -> anyhow::Result<Kek> {
let arr: [u8; 32] = bytes.as_slice().try_into().map_err(|_| anyhow!("{} is {} bytes, not 32", what.display(), bytes.len()))?;
Ok(Kek::from_bytes(arr))
};
let old = key32(std::fs::read(&path).with_context(|| format!("reading the replayed KEK {}", path.display()))?, &path)?;
let next_path = control_dir.join(KEK_NEXT);
let env = Kek::from_env()?;
let new = match &env {
Some(k) => k.clone(),
None => match std::fs::read(&next_path) {
Ok(bytes) => key32(bytes, &next_path)?,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
let k = Kek::generate()?;
write_key_file(control_dir, KEK_NEXT, &hex::decode(k.to_hex())?)?;
k
}
Err(e) => return Err(e).with_context(|| format!("reading {}", next_path.display())),
},
};
let mut moved = 0usize;
let mut dropped = 0usize;
if new.to_hex() != old.to_hex() {
log.append(Kind::Kek, &Meta::Kek {}, &hex::decode(new.to_hex())?).context("framing the restored box's own KEK")?;
for (key, row) in rows {
let Some(sealed) = row else { continue };
if !gunnar_core::is_sealed(sealed) || gunnar_core::unseal(&new, key, sealed).is_ok() {
continue;
}
let plain = match gunnar_core::unseal(&old, key, sealed) {
Ok(plain) => plain,
Err(err) => {
tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
key = %key,
%err,
"a secret that opens under NEITHER the replayed KEK nor this box's own: it \
was written by this box before the restore, under a key that is gone. \
DROPPED — it is not the restored primary's and it cannot be read by \
anyone. If something needs it, it will be minted again on first use"
);
raw.delete(key).with_context(|| format!("dropping the unopenable {key}"))?;
dropped += 1;
continue;
}
};
let resealed = gunnar_core::seal(&new, key, &plain)?;
log.append(Kind::Secret, &Meta::Secret { key: key.clone(), deleted: false }, &resealed)
.with_context(|| format!("framing the re-sealed {key}"))?;
raw.put(key, &resealed).with_context(|| format!("storing the re-sealed {key}"))?;
moved += 1;
}
}
if env.is_some() {
match std::fs::remove_file(&path) {
Ok(()) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => return Err(e).with_context(|| format!("removing the replayed KEK {}", path.display())),
}
} else {
std::fs::rename(&next_path, &path).with_context(|| format!("installing the new KEK at {}", path.display()))?;
}
crate::durable::sync_dir(control_dir).context("flushing the re-sealed KEK into place")?;
crate::durable::remove_durable(root, RESEAL)?;
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, moved, dropped, "a restored box re-sealed its secrets under a KEK of its own; the replayed KEK is gone from this disk");
Ok(new)
}
fn write_key_file(dir: &Path, name: &str, bytes: &[u8]) -> anyhow::Result<()> {
use std::io::Write as _;
std::fs::create_dir_all(dir)?;
let tmp = dir.join(format!("{name}.tmp"));
let mut options = std::fs::OpenOptions::new();
options.write(true).create(true).truncate(true);
{
use std::os::unix::fs::OpenOptionsExt as _;
options.mode(0o600);
}
let mut f = options.open(&tmp)?;
f.write_all(bytes)?;
f.sync_all()?;
drop(f);
std::fs::rename(&tmp, dir.join(name))?;
crate::durable::sync_dir(dir)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Progress {
Applied { count: u64, head: u64 },
CaughtUp { head: u64 },
SnapshotNeeded { wanted: u64, first: u64 },
SnapshotInstalled { at: u64 },
Diverged {
own: u64,
primary_head: u64,
},
}
pub const FAILOVER_AFTER: u32 = 5;
pub struct Follower {
stores: Arc<ServerStores>,
source: Box<dyn LogSource>,
fetch_bytes: u64,
reapply: AtomicBool,
failures: Arc<Mutex<Option<Failing>>>,
said_peak: std::sync::atomic::AtomicU64,
marked: std::sync::atomic::AtomicU64,
volume: Arc<dyn snapshot::Volume>,
}
pub fn own_memory() -> Option<(u64, u64)> {
let s = std::fs::read_to_string("/proc/self/status").ok()?;
let kib = |key: &str| s.lines().find_map(|l| l.strip_prefix(key)).and_then(|v| v.split_whitespace().next()).and_then(|v| v.parse::<u64>().ok());
Some((kib("VmHWM:")? << 10, kib("VmRSS:")? << 10))
}
pub const SAY_MEMORY_STEP: u64 = 32 << 20;
pub fn memory_is_news(said: u64, peak: u64) -> bool {
said == 0 || peak >= said.saturating_add(SAY_MEMORY_STEP)
}
#[derive(Debug, Clone)]
pub struct Failing {
shape: String,
since: std::time::Instant,
said: std::time::Instant,
polls: u64,
}
pub const RESAY_FAILING: Duration = Duration::from_secs(300);
pub(crate) fn refused_for_skew(err: &anyhow::Error) -> bool {
format!("{err:#}").contains(gunnar_grpc::auth::CLOCK_SKEW_PHRASE)
}
fn failure_shape(err: &anyhow::Error) -> String {
let text = format!("{err:#}");
let mut out = String::with_capacity(text.len().min(200));
let mut in_num = false;
for c in text.chars().take(200) {
if c.is_ascii_digit() {
if !in_num {
out.push('#');
}
in_num = true;
} else {
in_num = false;
out.push(c);
}
}
out
}
struct Appender<'a> {
stores: &'a ServerStores,
reapply: &'a AtomicBool,
open: Option<(txlog::Pending<'a>, u16, Vec<u8>, u64)>,
next: u64,
count: u64,
group: Option<Group>,
marked: &'a std::sync::atomic::AtomicU64,
}
struct Group {
entries: Vec<(u64, u16, Vec<u8>)>,
bytes: u64,
since: std::time::Instant,
}
impl Group {
fn new() -> Self {
Group { entries: Vec::new(), bytes: 0, since: std::time::Instant::now() }
}
fn full(&self) -> bool {
self.entries.len() >= GROUP_ENTRIES || self.bytes >= GROUP_BYTES || self.since.elapsed() >= GROUP_AGE
}
}
impl Appender<'_> {
fn flush(&mut self) -> anyhow::Result<()> {
drop(self.open.take());
let Some(g) = self.group.as_mut() else { return Ok(()) };
if g.entries.is_empty() {
return Ok(());
}
let entries = std::mem::take(&mut g.entries);
g.bytes = 0;
g.since = std::time::Instant::now();
self.stores.txlog().sync().context("making a group of entries durable in this twin's log")?;
self.reapply.store(true, Ordering::SeqCst);
take_group(self.stores, &entries, self.marked)?;
self.reapply.store(false, Ordering::SeqCst);
Ok(())
}
}
fn doc_name(kind: u16, meta: &[u8]) -> Option<String> {
if kind != Kind::ControlDoc as u16 {
return None;
}
match Meta::from_bytes(meta) {
Ok(Meta::ControlDoc { name, .. }) => Some(name),
_ => None,
}
}
fn fresh_names(root: &Path) -> anyhow::Result<std::collections::BTreeSet<String>> {
match std::fs::read(root.join(FRESH_DOCS)) {
Ok(b) => Ok(serde_json::from_slice(&b).with_context(|| format!("{} does not parse", root.join(FRESH_DOCS).display()))?),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Default::default()),
Err(e) => Err(e.into()),
}
}
fn take_group(stores: &ServerStores, entries: &[(u64, u16, Vec<u8>)], marked: &std::sync::atomic::AtomicU64) -> anyhow::Result<()> {
let db = stores.control_db();
let mut fresh = fresh_names(stores.root())?;
let mut done: Option<u64> = None;
db.begin_group()?;
struct AbortOnDrop<'a>(&'a gunnar_core::ControlDb);
impl Drop for AbortOnDrop<'_> {
fn drop(&mut self) {
self.0.abort_group();
}
}
let _abort = AbortOnDrop(&db);
let taken = (|| -> anyhow::Result<()> {
for (seq, kind, meta) in entries {
let (seq, kind) = (*seq, *kind);
let fresh_doc = if fresh.is_empty() { None } else { doc_name(kind, meta).filter(|n| fresh.contains(n)) };
let applied = if let Some(name) = fresh_doc {
if let Some(prev) = done {
db.commit_group(gunnar_core::TWIN_APPLIED, prev, false)?;
} else {
db.abort_group();
}
let a = take_logged(stores, seq, kind, meta)?;
db.set_mark(gunnar_core::TWIN_APPLIED, seq)?;
marked.store(seq, Ordering::SeqCst);
fresh.remove(&name);
db.begin_group()?;
a
} else {
take_logged(stores, seq, kind, meta)?
};
match applied {
Applied::Done | Applied::Stored => {}
Applied::Lost(what) => tracing::debug!(target: gunnar_core::observe::TARGET_TWIN, seq, %what, "replayed a lost CAS"),
Applied::Unknown(kind) => tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, seq, kind, "kept a frame of a kind this build does not apply"),
}
done = Some(seq);
}
Ok(())
})();
match (taken, done) {
(Ok(()), Some(last)) => {
let force = last >= marked.load(Ordering::SeqCst).saturating_add(MARK_EVERY);
if db.commit_group(gunnar_core::TWIN_APPLIED, last, force).context("committing the group's control-plane writes")? {
marked.store(last, Ordering::SeqCst);
}
Ok(())
}
(Ok(()), None) => {
db.abort_group();
Ok(())
}
(Err(err), _) => {
db.abort_group();
Err(err)
}
}
}
impl FrameSink for Appender<'_> {
fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()> {
if let Some((p, ..)) = &self.open {
anyhow::bail!("the primary began seq {seq} with seq {} unfinished", p.seq());
}
if seq != self.next {
return Err(anyhow!("the primary sent seq {seq} where {} was due", self.next));
}
let pending = self.stores.txlog().begin(Some(seq), kind, meta, blob_len)?;
self.open = Some((pending, kind, meta.to_vec(), blob_len));
Ok(())
}
fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()> {
let (p, ..) = self.open.as_mut().ok_or_else(|| anyhow!("the primary sent blob bytes outside an entry"))?;
let seq = p.seq();
p.write(bytes).with_context(|| format!("seq {seq}: writing its blob into this box's log"))?;
Ok(())
}
fn end(&mut self) -> anyhow::Result<()> {
let (p, kind, meta, blob_len) = self.open.take().ok_or_else(|| anyhow!("the primary ended an entry it never began"))?;
if let Some(g) = self.group.as_mut() {
let seq = p.commit_unsynced()?;
if g.entries.is_empty() {
g.since = std::time::Instant::now();
}
g.bytes += blob_len + meta.len() as u64;
g.entries.push((seq, kind, meta));
let full = g.full();
self.next = seq + 1;
self.count += 1;
if full {
self.flush()?;
}
return Ok(());
}
let seq = p.commit()?;
self.reapply.store(true, Ordering::SeqCst);
let applied = take_logged(self.stores, seq, kind, &meta)?;
self.reapply.store(false, Ordering::SeqCst);
match applied {
Applied::Done => {}
Applied::Stored => {}
Applied::Lost(what) => tracing::debug!(target: gunnar_core::observe::TARGET_TWIN, seq, %what, "replayed a lost CAS"),
Applied::Unknown(kind) => tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, seq, kind, "kept a frame of a kind this build does not apply"),
}
self.next = seq + 1;
self.count += 1;
Ok(())
}
}
fn resyncing_why(first: u64, own: u64, dropped: Option<&snapshot::Dropped>) -> String {
match dropped {
Some(d) => format!("taking a full copy from its primary (it retains from {first}) — {}", d.sentence()),
None => format!(
"taking a full copy from its primary (it retains from {first}; this twin holds up to {own}) — its previous \
copy is kept until the new one is whole, and it is NOT a current copy until then"
),
}
}
fn give_up_stale_copy(stores: &ServerStores, short: &snapshot::Short, first: u64, primary_head: u64, answered: i64) -> anyhow::Result<u64> {
let root = stores.root();
let log = stores.txlog();
let head = log.head_seq();
let base_at = snapshot::base_at(root).ok().flatten();
let freed = snapshot::bytes_under(&root.join(snapshot::BASE)) + snapshot::bytes_under(&root.join(format!("{}.old", snapshot::BASE))) + log.bytes();
let d = snapshot::Dropped { at_unix_ms: now_ms(), head, base_at, freed };
snapshot::mark_dropped(root, &d)?;
tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
event = "twin.resync_without_copy",
head,
freed,
staged = short.staged,
free = short.free,
total = short.total,
"ALERT the full resync does not fit beside this twin's stale copy ({} staged, {} free of {}{}): {}",
human(short.staged),
human(short.free),
human(short.total),
if short.enospc { ", the volume said ENOSPC" } else { "" },
d.sentence()
);
finish_giving_up(stores)?;
set_status(root, TwinStatus {
applied_seq: head,
primary_head_seq: primary_head,
state: State::Resyncing,
why: resyncing_why(first, head, Some(&d)),
last_answer_unix_ms: answered,
last_level_unix_ms: 0,
});
Ok(freed)
}
fn finish_giving_up(stores: &ServerStores) -> anyhow::Result<()> {
let root = stores.root();
for dir in [root.join(snapshot::BASE), root.join(format!("{}.old", snapshot::BASE))] {
match std::fs::remove_dir_all(&dir) {
Ok(()) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => return Err(e).with_context(|| format!("giving up {}", dir.display())),
}
}
let log = stores.txlog();
let head = log.head_seq();
if log.first_seq() <= head || log.bytes() > 0 {
log.restart_after(head).context("giving up the stale log")?;
}
Ok(())
}
impl Follower {
pub fn new(stores: Arc<ServerStores>, source: Box<dyn LogSource>) -> Self {
Self {
stores,
source,
fetch_bytes: DEFAULT_FETCH_BYTES,
reapply: AtomicBool::new(true),
failures: Arc::new(Mutex::new(None)),
said_peak: std::sync::atomic::AtomicU64::new(0),
marked: std::sync::atomic::AtomicU64::new(0),
volume: Arc::new(snapshot::Disk),
}
}
pub fn with_volume(mut self, volume: Arc<dyn snapshot::Volume>) -> Self {
self.volume = volume;
self
}
pub fn sharing_failures(mut self, failures: Arc<Mutex<Option<Failing>>>) -> Self {
self.failures = failures;
self
}
fn say_memory(&self, head: u64) {
let Some((peak, now)) = own_memory() else { return };
let said = self.said_peak.load(Ordering::Relaxed);
if !memory_is_news(said, peak) {
return;
}
self.said_peak.store(peak, Ordering::Relaxed);
tracing::info!(
target: gunnar_core::observe::TARGET_TWIN,
peak_mib = peak >> 20,
rss_mib = now >> 20,
head,
"twin memory: peak resident set {} MiB, {} MiB now, following through seq {head}",
peak >> 20,
now >> 20
);
}
fn failures_ended(&self) {
if let Some(f) = self.failures.lock().unwrap_or_else(|p| p.into_inner()).take() {
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, polls = f.polls, seconds = f.since.elapsed().as_secs(), "the primary answers again: following on after {} failed poll(s) over {} s", f.polls, f.since.elapsed().as_secs());
}
}
fn failed(&self, err: &anyhow::Error) {
if refused_for_skew(err) {
estate_clock::trigger::skew_refused(0);
}
let shape = failure_shape(err);
let now = std::time::Instant::now();
let mut g = self.failures.lock().unwrap_or_else(|p| p.into_inner());
match g.as_mut() {
Some(f) if f.shape == shape => {
f.polls += 1;
if now.duration_since(f.said) >= RESAY_FAILING {
f.said = now;
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, polls = f.polls, seconds = now.duration_since(f.since).as_secs(), "poll still failing the same way; this twin keeps asking");
} else {
tracing::debug!(target: gunnar_core::observe::TARGET_TWIN, %err, polls = f.polls, "poll failed (said already)");
}
}
_ => {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "poll failed; retrying (said once while it fails this way)");
*g = Some(Failing { shape, since: now, said: now, polls: 1 });
}
}
}
pub fn with_fetch_bytes(mut self, bytes: u64) -> Self {
self.fetch_bytes = bytes.max(1);
self
}
pub fn applied_seq(&self) -> u64 {
self.stores.txlog().head_seq()
}
fn reapply_head(&self) -> anyhow::Result<()> {
if !self.reapply.load(Ordering::SeqCst) {
return Ok(());
}
if self.stores.is_twin() {
return self.retake_from_marker();
}
let log = self.stores.txlog();
let head = log.head_seq();
if head > 0 && head >= log.first_seq() {
let at = log.locate_blob(head)?;
take_logged(&self.stores, head, at.kind, &at.meta).with_context(|| format!("re-applying seq {head}, the head of this twin's log, before fetching anything newer"))?;
}
self.reapply.store(false, Ordering::SeqCst);
Ok(())
}
fn retake_from_marker(&self) -> anyhow::Result<()> {
let log = self.stores.txlog();
let db = self.stores.control_db();
let head = log.head_seq();
match db.mark(gunnar_core::TWIN_APPLIED)? {
Some(m) if m >= head => {
if m > head {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, marker = m, head, "the applied marker is past this twin's log head; following on from the head");
db.set_mark(gunnar_core::TWIN_APPLIED, head)?;
}
self.marked.store(head, Ordering::SeqCst);
}
Some(m) => {
self.marked.store(m, Ordering::SeqCst);
let from = (m + 1).max(log.first_seq());
let mut entries = Vec::with_capacity((head + 1).saturating_sub(from).min(1 << 16) as usize);
for seq in from..=head {
let at = log.locate_blob(seq)?;
entries.push((seq, at.kind, at.meta));
}
take_group(&self.stores, &entries, &self.marked).with_context(|| format!("re-taking seq {from}..={head}, past this twin's applied marker {m}, before fetching anything newer"))?;
if self.marked.load(Ordering::SeqCst) < head {
db.set_mark(gunnar_core::TWIN_APPLIED, head)?;
self.marked.store(head, Ordering::SeqCst);
}
}
None => {
if head > 0 && head >= log.first_seq() {
let at = log.locate_blob(head)?;
take_logged(&self.stores, head, at.kind, &at.meta).with_context(|| format!("re-applying seq {head}, the head of this twin's log, before fetching anything newer"))?;
}
db.set_mark(gunnar_core::TWIN_APPLIED, head)?;
self.marked.store(head, Ordering::SeqCst);
}
}
self.reapply.store(false, Ordering::SeqCst);
Ok(())
}
pub fn step(&self) -> anyhow::Result<Progress> {
let mut said_why = false;
let mut heard: Option<(u64, i64)> = None;
let r = self.step_inner(&mut said_why, &mut heard);
if let (Err(err), false) = (&r, said_why) {
let last = status(self.stores.root());
let (primary_head_seq, last_answer_unix_ms) = heard.unwrap_or((
last.as_ref().map(|s| s.primary_head_seq).unwrap_or(0),
last.as_ref().map(|s| s.last_answer_unix_ms).unwrap_or(0),
));
let said = format!("{err:#}");
let no_copy = snapshot::dropped(self.stores.root()).map(|d| format!(" — {}", d.sentence())).unwrap_or_default();
if let Some(at) = said.find(gunnar_grpc::TWIN_TOO_SMALL) {
set_status(self.stores.root(), TwinStatus {
applied_seq: self.applied_seq(),
primary_head_seq,
state: State::Recovering,
why: format!("{}{no_copy}", &said[at..]),
last_answer_unix_ms,
last_level_unix_ms: 0,
});
return r;
}
let holds = if no_copy.is_empty() { "what it holds stops here".to_owned() } else { format!("it holds NO copy{no_copy}") };
set_status(self.stores.root(), TwinStatus {
applied_seq: self.applied_seq(),
primary_head_seq,
state: State::Stalled,
why: format!("this twin could not take what its primary sent — it is NOT a current copy, and {holds}: {err:#}"),
last_answer_unix_ms,
last_level_unix_ms: 0,
});
}
r
}
fn step_inner(&self, said_why: &mut bool, heard: &mut Option<(u64, i64)>) -> anyhow::Result<Progress> {
let installing = snapshot::installing(self.stores.root());
if !installing {
self.reapply_head()?;
}
let (first, primary_head, their_id, their_generation) = match self.source.head_identity_generation() {
Ok(h) => h,
Err(err) => {
let last = status(self.stores.root());
set_status(self.stores.root(), TwinStatus {
applied_seq: self.applied_seq(),
primary_head_seq: last.as_ref().map(|s| s.primary_head_seq).unwrap_or(0),
state: State::Unreachable,
why: err.to_string(),
last_answer_unix_ms: last.as_ref().map(|s| s.last_answer_unix_ms).unwrap_or(0),
last_level_unix_ms: 0,
});
*said_why = true;
return Err(err);
}
};
let answered = now_ms();
let own = self.applied_seq();
*heard = Some((primary_head, answered));
if let Some(theirs) = their_id.as_deref().map(str::trim).filter(|s| !s.is_empty()) {
match log_id(self.stores.root()) {
Some(mine) if mine != theirs => {
let why = format!(
"the box at the other end is log {theirs}; this twin holds log {mine}. Two different histories, \
whatever their sequence numbers say — this is a re-imaged or reinstalled primary at a familiar \
address, or an address that now points at another set. NOTHING is applied on top of it, and \
retrying cannot fix it: either point this twin at the box that holds {mine}, or — if {mine} is \
gone — restore a primary FROM this twin, which keeps the id, and it will follow again."
);
set_status(self.stores.root(), TwinStatus {
applied_seq: own,
primary_head_seq: primary_head,
state: State::Diverged,
why,
last_answer_unix_ms: answered,
last_level_unix_ms: 0,
});
return Ok(Progress::Diverged { own, primary_head });
}
Some(_) => {}
None => {
pin_log_id(self.stores.root(), theirs).context("pinning the primary's log identity on this twin")?;
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, log_id = %theirs, "this twin now knows which log it follows");
}
}
}
let pinned = pinned_generation(self.stores.root())?;
if their_generation < pinned {
let why = format!(
"the box at the other end is generation {their_generation} of this log; this twin already follows generation \
{pinned}, a later restore of it. This is the OLD main come back after its replacement was restored: the same \
log id and the same key, so nothing else can tell them apart. NOTHING is applied from it — its writes since \
the restore belong to no history; this poll handed it the proof, so it fences itself ({LOG_FENCED}). Point \
this twin back at the restored main, and retire the old box."
);
set_status(self.stores.root(), TwinStatus {
applied_seq: own,
primary_head_seq: primary_head,
state: State::Diverged,
why,
last_answer_unix_ms: answered,
last_level_unix_ms: 0,
});
return Ok(Progress::Diverged { own, primary_head });
}
if !installing && primary_head < own {
let why = format!(
"the primary answers head {primary_head}, BELOW the {own} this twin has already applied. A log's head never goes \
backwards, so this is NOT the log this twin was following (a re-imaged or restored primary, or an address that now \
points at a different set). Nothing is applied on top of it; an operator decides whether to re-join from scratch."
);
set_status(self.stores.root(), TwinStatus {
applied_seq: own,
primary_head_seq: primary_head,
state: State::Diverged,
why,
last_answer_unix_ms: answered,
last_level_unix_ms: 0,
});
return Ok(Progress::Diverged { own, primary_head });
}
if their_generation > pinned {
pin_generation(self.stores.root(), their_generation).context("pinning the primary's log generation on this twin")?;
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, generation = their_generation, "this twin now follows generation {their_generation} of its log and refuses any older one");
}
let wanted = own + 1;
if installing || (wanted <= primary_head && first > wanted) {
let root = self.stores.root();
let twin = self.stores.is_twin();
let mut given_up = false;
if twin {
if let Some(was) = snapshot::too_small_total(root) {
if self.volume.space(root).is_some_and(|(total, _)| total <= was) {
let no_copy = snapshot::dropped(root).map(|d| format!(" — {}", d.sentence())).unwrap_or_default();
anyhow::bail!(
"{} this twin's last full resync ran out of room on its {} volume with its stale copy given up; it asks main for another only once its volume is larger (main holds its writes for every snapshot stream){no_copy}",
gunnar_grpc::TWIN_TOO_SMALL,
human(was)
);
}
}
if snapshot::dropped(root).is_some() {
finish_giving_up(&self.stores)?;
given_up = true;
}
set_status(root, TwinStatus {
applied_seq: own,
primary_head_seq: primary_head,
state: State::Resyncing,
why: resyncing_why(first, own, snapshot::dropped(root).as_ref()),
last_answer_unix_ms: answered,
last_level_unix_ms: 0,
});
}
let mut stager = snapshot::Stager::begin_on(root, Arc::clone(&self.volume))?;
if twin {
let stores = Arc::clone(&self.stores);
stager = stager.making_room(Box::new(move |short: &snapshot::Short| give_up_stale_copy(&stores, short, first, primary_head, answered)), given_up);
}
let fetched = self.source.snapshot(&mut stager);
if let Err(err) = &fetched {
if let Some(t) = err.downcast_ref::<snapshot::TooSmall>() {
snapshot::note_too_small(root, t)?;
}
}
if fetched.context("fetching a snapshot to close the hole")? {
let snap = stager.finish()?;
let was_given_up = snapshot::dropped(root).is_some();
snapshot::install(&self.stores, &snap)?;
self.stores.control_db().set_mark(gunnar_core::TWIN_APPLIED, snap.at)?;
self.marked.store(snap.at, Ordering::SeqCst);
self.reapply.store(false, Ordering::SeqCst);
let whole_again = if was_given_up { "; the stale copy given up for it is replaced — this twin holds a whole copy again" } else { "" };
set_status(root, TwinStatus {
applied_seq: snap.at,
primary_head_seq: primary_head.max(snap.at),
state: State::Following,
why: format!("installed a snapshot at {} (the source retained from {first}){whole_again}", snap.at),
last_answer_unix_ms: answered,
last_level_unix_ms: 0,
});
return Ok(Progress::SnapshotInstalled { at: snap.at });
}
}
if installing {
let why = match snapshot::dropped(self.stores.root()) {
Some(d) => format!(
"{} — and the source cannot make a snapshot to replace it; nothing is applied on \
top of an empty copy",
d.sentence()
),
None => format!(
"this box crashed inside a snapshot install ({}/{}) and the source cannot make a \
snapshot to replace it; half a snapshot is not a state to follow on from, so \
nothing is applied on top of it",
self.stores.root().display(),
snapshot::INSTALLING
),
};
set_status(self.stores.root(), TwinStatus {
applied_seq: own,
primary_head_seq: primary_head,
state: State::SnapshotNeeded,
why: why.clone(),
last_answer_unix_ms: answered,
last_level_unix_ms: 0,
});
*said_why = true;
return Err(anyhow!(why));
}
if wanted <= primary_head && first > wanted {
set_status(self.stores.root(), TwinStatus {
applied_seq: own,
primary_head_seq: primary_head,
state: State::SnapshotNeeded,
why: format!("the primary retains from {first}; this twin holds up to {own} and needs {wanted} next"),
last_answer_unix_ms: answered,
last_level_unix_ms: 0,
});
return Ok(Progress::SnapshotNeeded { wanted, first });
}
let mut count = 0u64;
let mut next = wanted;
while next <= primary_head {
let group = self.stores.is_twin().then(Group::new);
let mut into = Appender { stores: &self.stores, reapply: &self.reapply, open: None, next, count: 0, group, marked: &self.marked };
let fetched = self.source.fetch_into(next, primary_head, self.fetch_bytes, &mut into);
let flushed = into.flush();
let (taken, after) = (into.count, into.next);
drop(into);
count += taken;
fetched?;
flushed?;
if after == next {
return Err(anyhow!("the primary answered an empty range for {next}..={primary_head}"));
}
next = after;
}
let head = self.applied_seq();
set_status(self.stores.root(), TwinStatus {
applied_seq: head,
primary_head_seq: primary_head,
state: State::Following,
why: String::new(),
last_answer_unix_ms: answered,
last_level_unix_ms: 0,
});
Ok(if count == 0 { Progress::CaughtUp { head } } else { Progress::Applied { count, head } })
}
pub fn run(&self, poll: Duration, stop: &AtomicBool) -> Ended {
let hold = maintenance::hold_of(self.stores.root());
let mut failed_in_a_row = 0u32;
let mut primary_growing = false;
let mut recovering = false;
while !stop.load(Ordering::Relaxed) {
let busy = hold.busy.lock().unwrap_or_else(|p| p.into_inner());
if hold.held.load(Ordering::SeqCst) {
drop(busy);
self.maintenance_beat();
nap(poll, stop);
continue;
}
let stepped = self.step();
drop(busy);
let mut again = false;
match stepped {
Ok(Progress::SnapshotNeeded { wanted, first }) => {
tracing::error!(target: gunnar_core::observe::TARGET_TWIN, wanted, first, "snapshot needed — the twin stops rather than append past a hole");
return Ended::SnapshotNeeded;
}
Ok(Progress::Diverged { own, primary_head }) => {
tracing::error!(target: gunnar_core::observe::TARGET_TWIN, own, primary_head, "DIVERGED — the primary's head is below this twin's; the twin stops rather than pretend to follow a different log");
return Ended::Diverged;
}
Ok(p) => {
failed_in_a_row = 0;
self.failures_ended();
again = matches!(p, Progress::Applied { .. } | Progress::SnapshotInstalled { .. });
if let Progress::Applied { head, .. } = p {
self.say_memory(head);
}
if primary_growing {
primary_growing = false;
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, "the primary's data set is back: following on");
}
if recovering {
recovering = false;
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, "this twin's volume holds main's set again: following on");
}
}
Err(err) if format!("{err:#}").contains(gunnar_grpc::PRIMARY_GROWING) => {
failed_in_a_row = 0;
if !primary_growing {
primary_growing = true;
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, "the primary is growing its data set: this twin waits and follows on after the remount");
}
for _ in 0..GROWING_WAIT_POLLS {
if stop.load(Ordering::Relaxed) {
break;
}
nap(poll, stop);
}
continue;
}
Err(err) if format!("{err:#}").contains(gunnar_grpc::TWIN_TOO_SMALL) => {
failed_in_a_row = 0;
if !recovering {
recovering = true;
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "recovering: the primary will not let this twin follow until its volume holds main's grown set; it keeps polling and follows once grown");
}
for _ in 0..GROWING_WAIT_POLLS {
if stop.load(Ordering::Relaxed) {
break;
}
nap(poll, stop);
}
continue;
}
Err(err) if format!("{err:#}").contains(STALLED) => {
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, %err, "the primary said nothing on this connection: re-dialling it now");
return Ended::Redial;
}
Err(err) => {
failed_in_a_row += 1;
self.failed(&err);
if self.source.alternatives() && failed_in_a_row >= FAILOVER_AFTER {
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, failed_in_a_row, "re-dialling the primary on the address that answers (tunnel or utility network)");
return Ended::Redial;
}
}
}
if !again {
nap(poll, stop);
}
}
Ended::Stopped
}
pub fn maintenance_beat(&self) {
let root = self.stores.root();
let last = status(root);
let mut s = TwinStatus {
applied_seq: self.applied_seq(),
primary_head_seq: last.as_ref().map(|s| s.primary_head_seq).unwrap_or(0),
state: State::Maintenance,
why: maintenance::WHY.to_owned(),
last_answer_unix_ms: last.as_ref().map(|s| s.last_answer_unix_ms).unwrap_or(0),
last_level_unix_ms: 0,
};
set_status(root, s.clone());
if let Ok((_, head)) = self.source.head() {
s.primary_head_seq = head;
s.last_answer_unix_ms = now_ms();
set_status(root, s);
}
}
}
pub mod maintenance {
use super::*;
pub const WHY: &str = "this twin is growing its own data set (the same runtime growth as main: drained, \
unmounted, grown, mounted back — never restarted); it applies nothing until the set \
is back, then follows on from its own head";
pub(super) struct Hold {
pub(super) held: AtomicBool,
pub(super) busy: Mutex<()>,
}
static HOLDS: OnceLock<Mutex<HashMap<PathBuf, Arc<Hold>>>> = OnceLock::new();
pub(super) fn hold_of(root: &Path) -> Arc<Hold> {
let mut m = HOLDS.get_or_init(|| Mutex::new(HashMap::new())).lock().unwrap_or_else(|p| p.into_inner());
Arc::clone(m.entry(root.to_path_buf()).or_insert_with(|| Arc::new(Hold { held: AtomicBool::new(false), busy: Mutex::new(()) })))
}
pub fn hold(root: &Path) {
let h = hold_of(root);
h.held.store(true, Ordering::SeqCst);
drop(h.busy.lock().unwrap_or_else(|p| p.into_inner()));
let last = status(root);
set_status(root, TwinStatus {
applied_seq: last.as_ref().map(|s| s.applied_seq).unwrap_or(0),
primary_head_seq: last.as_ref().map(|s| s.primary_head_seq).unwrap_or(0),
state: State::Maintenance,
why: WHY.to_owned(),
last_answer_unix_ms: last.as_ref().map(|s| s.last_answer_unix_ms).unwrap_or(0),
last_level_unix_ms: 0,
});
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, root = %root.display(), "maintenance: the twin's follower is held for its own data-set growth");
}
pub fn release(root: &Path) {
hold_of(root).held.store(false, Ordering::SeqCst);
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, root = %root.display(), "maintenance over: the twin follows again");
}
pub fn is_held(root: &Path) -> bool {
hold_of(root).held.load(Ordering::SeqCst)
}
#[cfg(test)]
mod tests;
}
struct OffThreadRuntime(Option<tokio::runtime::Runtime>);
impl OffThreadRuntime {
fn new(rt: tokio::runtime::Runtime) -> Self {
OffThreadRuntime(Some(rt))
}
fn get(&self) -> &tokio::runtime::Runtime {
self.0.as_ref().expect("the follower's runtime is taken only by Drop")
}
fn block_on<F: std::future::Future>(&self, f: F) -> F::Output {
self.get().block_on(f)
}
}
impl Drop for OffThreadRuntime {
fn drop(&mut self) {
let Some(rt) = self.0.take() else { return };
match std::thread::Builder::new().name("twin-rt-drop".to_string()).spawn(move || drop(rt)) {
Ok(_) => {}
Err(e) => {
tracing::error!(
target: gunnar_core::observe::TARGET_TWIN, error = %e,
"the follower's runtime could not be dropped on a thread of its own; it is leaked rather than dropped here"
);
}
}
}
}
pub struct GrpcSource {
rt: OffThreadRuntime,
client: Mutex<gunnar_proto::mirror_client::MirrorClient<gunnar_grpc::client_signer::ControlChannel>>,
signer: gunnar_core::ssh_key::PrivateKey,
reporter: Option<Box<dyn Fn() -> gunnar_proto::TwinReport + Send + Sync>>,
root: std::path::PathBuf,
alternatives: bool,
}
fn say(status: impl std::fmt::Display) -> anyhow::Error {
anyhow!("{status}")
}
impl GrpcSource {
pub fn connect(at: &str, signer: gunnar_core::ssh_key::PrivateKey, root: &Path) -> anyhow::Result<Self> {
let rt = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
let loopback = gunnar_grpc::client_signer::is_loopback_target(at);
let mut alternatives = false;
let (trust, dial, domain) = match crate::twin_tls::load(root)? {
Some((trust, port)) => {
let candidates = crate::twin_tls::load_host(root)
.map(|r| crate::twin_tls::hosts(&r))
.filter(|h| !h.is_empty())
.unwrap_or_else(|| vec![gunnar_grpc::client_signer::tls_domain_for(at)]);
alternatives = candidates.len() > 1;
let dial = pick_replication_dial(&candidates, port, DIAL_PROBE);
(Some(trust), dial, gunnar_grpc::client_signer::REPLICATION_SERVER_NAME.to_owned())
}
None => (
gunnar_grpc::client_signer::ClientTrust::from_env().map_err(|e| anyhow!(e))?,
at.to_owned(),
gunnar_grpc::client_signer::tls_domain_for(at),
),
};
let channel = {
let _g = rt.enter();
match (loopback, &trust) {
(_, Some(trust)) => {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0);
match gunnar_core::expiry::grade_pem(now, &trust.cert) {
Ok(expiry) => {
let line = expiry.line("this twin's client certificate");
if expiry.is_refusal() {
anyhow::bail!("{line}");
}
match expiry.grade {
gunnar_core::expiry::Grade::Renew => {
tracing::error!(target: gunnar_core::observe::TARGET_TWIN, days_left = expiry.days_left, "{line}")
}
_ => tracing::info!(target: gunnar_core::observe::TARGET_TWIN, days_left = expiry.days_left, "{line}"),
}
}
Err(err) => anyhow::bail!(
"this twin's client certificate ({}) could not be read, so nothing \
can say how long it is good for: {err}",
gunnar_grpc::client_signer::CLIENT_CERT_ENV
),
}
tracing::info!(
target: gunnar_core::observe::TARGET_TWIN,
%at, %dial, %domain, suite = gunnar_grpc::preferred_bulk_suite(),
"following the primary over MUTUAL TLS 1.3"
);
crate::narrate::at(
crate::narrate::Stage::Follow,
&format!(
"primary {at} dialled as {dial} (server name {domain}) over mutual \
TLS 1.3; this twin's client certificate is held and in grade"
),
);
gunnar_grpc::client_signer::mutual_channel_with(&dial, trust, &domain, kept_alive()).map_err(|e| anyhow!(e))?
}
(true, None) => {
tracing::info!(
target: gunnar_core::observe::TARGET_TWIN, %at,
"following a LOOPBACK primary in the clear: the bytes never leave this box"
);
gunnar_grpc::client_signer::plain_channel_with(at, kept_alive()).map_err(|e| anyhow!(e))?
}
(false, None) => return Err(anyhow::Error::new(AwaitingCertificate).context(format!(
"this twin's primary is {at}, which is another machine, and no client \
certificate is configured on this box — so the only way to follow it would \
be in the clear. It will not. The log this stream carries holds the \
estate's sealed secrets and the KEK as seq 1.\n\n\
Set all three of {}, {} and {} to the files this box was enrolled with, and \
make sure the primary is reached over the WireGuard address its listener \
bound (the certificate carries that address, not the tunnel's name).",
gunnar_grpc::client_signer::CLIENT_CERT_ENV,
gunnar_grpc::client_signer::CLIENT_KEY_ENV,
gunnar_grpc::client_signer::CA_ENV,
))),
}
};
Ok(Self {
rt: OffThreadRuntime::new(rt),
client: Mutex::new(gunnar_proto::mirror_client::MirrorClient::new(channel)),
signer,
reporter: None,
root: root.to_path_buf(),
alternatives,
})
}
}
pub const HEAD_WITHIN: Duration = Duration::from_secs(3);
pub const FETCH_IDLE_WITHIN: Duration = Duration::from_secs(15);
pub const SNAPSHOT_IDLE_WITHIN: Duration = Duration::from_secs(60);
pub const KEEPALIVE_EVERY: Duration = Duration::from_secs(10);
pub const KEEPALIVE_WITHIN: Duration = Duration::from_secs(20);
pub const STALLED: &str = "[twin-primary-stalled]";
fn kept_alive() -> impl FnOnce(gunnar_grpc::client_signer::Endpoint) -> gunnar_grpc::client_signer::Endpoint {
gunnar_grpc::client_signer::keep_alive(KEEPALIVE_EVERY, KEEPALIVE_WITHIN)
}
pub const DIAL_PROBE: Duration = Duration::from_secs(3);
fn dial_of(host: &str, port: u16) -> String {
if host.contains(':') { format!("[{host}]:{port}") } else { format!("{host}:{port}") }
}
pub fn pick_replication_dial(candidates: &[String], port: u16, probe: Duration) -> String {
for host in candidates {
let dial = dial_of(host, port);
let Ok(addr) = dial.parse::<std::net::SocketAddr>() else { continue };
if std::net::TcpStream::connect_timeout(&addr, probe).is_ok() {
return dial;
}
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, %dial, "the primary's replication door does not answer here; trying its next address");
}
dial_of(candidates.first().map(String::as_str).unwrap_or("127.0.0.1"), port)
}
impl GrpcSource {
pub fn with_reporter(mut self, report: impl Fn() -> gunnar_proto::TwinReport + Send + Sync + 'static) -> Self {
self.reporter = Some(Box::new(report));
self
}
fn check_answer(&self, nonce: &str, head: &gunnar_proto::LogHeadResponse) -> anyhow::Result<()> {
let offered = head.answer_key.trim();
match pinned_primary_key(&self.root) {
Some(pinned) => {
anyhow::ensure!(
!head.answer_sig.trim().is_empty(),
"the box at this address answered UNSIGNED, and this twin has pinned the key its primary answers with \
({pinned}). An answer nobody signed is an answer from nobody: this twin applies nothing on it."
);
anyhow::ensure!(
offered == pinned,
"the box at this address now answers with key {offered}; this twin follows the box that answers with \
{pinned}. The address is the same and the box is not: nothing is applied. If the primary was rebuilt \
rather than restored, it is a different history too — re-join this twin deliberately, or restore a \
primary FROM it."
);
anyhow::ensure!(
gunnar_grpc::auth::verify_twin_answer_at(&pinned, &head.answer_sig, nonce, &head.log_id, head.first_seq, head.head_seq, head.generation),
"the answer from this address does not verify against the key this twin pinned ({pinned}) — the signature \
is not over this poll's challenge and these numbers (and this generation). A replayed or edited answer is not an answer."
);
Ok(())
}
None if offered.is_empty() => Ok(()),
None => {
anyhow::ensure!(
gunnar_grpc::auth::verify_twin_answer_at(offered, &head.answer_sig, nonce, &head.log_id, head.first_seq, head.head_seq, head.generation),
"the box at this address offered key {offered} with a signature that does not verify against it; this twin \
pins nothing it has not seen prove itself once"
);
pin_primary_key(&self.root, offered)
}
}
}
}
pub const TXLOG_SIZE_ALARM: &str = "txlog.size_over_cap";
pub const TXLOG_ALARM_ENV: &str = "GUNNAR_TXLOG_ALARM_BYTES";
pub const TXLOG_ALARM_DEFAULT_PERCENT: u64 = 25;
pub fn txlog_alarm_cap(env: Option<&str>, volume_total: u64) -> Result<Option<(u64, String)>, String> {
match env.map(str::trim).filter(|v| !v.is_empty()) {
Some(v) => v
.parse::<u64>()
.map(|b| Some((b, format!("{TXLOG_ALARM_ENV}={b}"))))
.map_err(|_| format!("{TXLOG_ALARM_ENV}={v:?} is not a number of bytes; the log's size is NOT watched")),
None if volume_total == 0 => Ok(None),
None => Ok(Some((
volume_total / 100 * TXLOG_ALARM_DEFAULT_PERCENT,
format!("the default, {TXLOG_ALARM_DEFAULT_PERCENT}% of the {volume_total}-byte data volume"),
))),
}
}
pub fn txlog_size_alarm(log_bytes: u64, cap: u64, how: &str) -> Option<gunnar_proto::HealthAlarm> {
(log_bytes > cap).then(|| gunnar_proto::HealthAlarm {
event: TXLOG_SIZE_ALARM.into(),
count: 1,
since_unix_ms: 0,
detail: format!(
"the transaction log holds {log_bytes} bytes on disk, over the cap of {cap} ({how}). \
Nothing has been deleted and nothing will be: let every twin catch up, then retain \
(`gunnar control twin retain --at <primary>`), or raise {TXLOG_ALARM_ENV}. A log this \
size needs {} bytes free on any box restored from it",
log_bytes.saturating_mul(RESTORE_BYTES_PER_LOG_BYTE)
),
})
}
pub fn measure_report(stores: &ServerStores) -> gunnar_proto::TwinReport {
let (total, used) = if maintenance::is_held(stores.root()) { (0, 0) } else { volume_bytes(stores.root()) };
let (cpu_millicores_busy, cpu_millicores_total) = own_load();
gunnar_proto::TwinReport {
applied_seq: stores.txlog().head_seq(),
disk_total_bytes: total,
disk_used_bytes: used,
state: status(stores.root()).map(|s| s.state.to_string()).unwrap_or_default(),
cpu_millicores_busy,
cpu_millicores_total,
}
}
pub fn own_head(stores: &ServerStores) -> gunnar_proto::LogHeadResponse {
let log = stores.txlog();
let mut head = gunnar_proto::LogHeadResponse {
head_seq: log.head_seq(),
first_seq: log.first_seq(),
bytes: log.bytes(),
log_id: crate::twin::log_id(stores.root()).unwrap_or_default(),
..Default::default()
};
if crate::twin::is_twin_root(stores.root()) {
head.twin = true;
if crate::twin::snapshot::not_whole(stores.root()).is_none() {
if let Ok(Some(at)) = crate::twin::snapshot::base_at(stores.root()) {
head.snapshot_at = at;
head.snapshot_bytes = crate::twin::snapshot::base_bytes(stores.root());
}
}
match crate::twin::status(stores.root()) {
Some(twin) => {
head.applied_seq = twin.applied_seq;
head.primary_head_seq = twin.primary_head_seq;
head.caught_up = twin.caught_up();
head.state = twin.state.to_string();
head.why = twin.why;
head.last_answer_unix_ms = twin.last_answer_unix_ms;
head.last_level_unix_ms = twin.last_level_unix_ms;
head.answer_ago_ms = if twin.last_answer_unix_ms > 0 {
crate::twin::now_ms().saturating_sub(twin.last_answer_unix_ms).max(0) as u64
} else {
0
};
}
None => {
head.applied_seq = log.head_seq();
head.state = "idle".to_owned();
head.why = "this box is a twin (twin.json is on its data root) but its follower has not \
reported since this process started: the primary's head, the lag and the age \
of this copy are all UNKNOWN"
.to_owned();
}
}
}
head
}
pub fn own_load() -> (Option<u64>, Option<u64>) {
let busy = crate::dashboard::probe::load1_value().ok().map(|l| (l * 1000.0).round().max(0.0) as u64);
let total = std::thread::available_parallelism().ok().map(|n| n.get() as u64 * 1000);
(busy, total)
}
pub fn volume_bytes(path: &Path) -> (u64, u64) {
match rustix::fs::statvfs(path) {
Ok(v) => {
let frsize = if v.f_frsize == 0 { v.f_bsize } else { v.f_frsize };
let total = v.f_blocks.saturating_mul(frsize);
let free = v.f_bfree.saturating_mul(frsize);
(total, total.saturating_sub(free))
}
Err(_) => (0, 0),
}
}
impl LogSource for GrpcSource {
fn alternatives(&self) -> bool {
self.alternatives
}
fn head(&self) -> anyhow::Result<(u64, u64)> {
let (first, head, _) = self.head_and_identity()?;
Ok((first, head))
}
fn head_and_identity(&self) -> anyhow::Result<(u64, u64, Option<String>)> {
let (first, head, id, _) = self.head_identity_generation()?;
Ok((first, head, id))
}
fn head_identity_generation(&self) -> anyhow::Result<(u64, u64, Option<String>, u64)> {
let mut client = self.client.lock().expect("twin client").clone();
let report = self.reporter.as_ref().map(|r| r());
let generation_proof = generation_proof(&self.root).unwrap_or_else(|err| {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "the kept generation proof does not read; relaying none");
None
});
let nonce = {
use gunnar_core::ssh_key::rand_core::Rng as _;
let mut bytes = [0u8; 16];
gunnar_core::os_rng().fill_bytes(&mut bytes);
bytes.iter().map(|b| format!("{b:02x}")).collect::<String>()
};
let req = gunnar_grpc::client_signer::authed(&self.signer, gunnar_proto::LogHeadRequest { report, nonce: nonce.clone(), generation_proof });
let head = self
.rt
.block_on(async { tokio::time::timeout(HEAD_WITHIN, client.log_head(req)).await })
.map_err(|_| anyhow!("{STALLED} the primary said nothing to Mirror.LogHead within {} s on this connection", HEAD_WITHIN.as_secs()))?
.map_err(say)?
.into_inner();
self.check_answer(&nonce, &head)?;
if let Err(err) = remember_generation_proof(&self.root, &nonce, &head) {
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "keeping the restored main's signed generation to relay it");
}
let id = Some(head.log_id).filter(|s| !s.trim().is_empty());
Ok((head.first_seq, head.head_seq, id, head.generation))
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
let mut whole = Gather::default();
self.fetch_into(from, to, max_bytes, &mut whole)?;
Ok(whole.out)
}
fn fetch_into(&self, from: u64, to: u64, max_bytes: u64, into: &mut dyn FrameSink) -> anyhow::Result<()> {
let mut client = self.client.lock().expect("twin client").clone();
let req = gunnar_grpc::client_signer::authed(&self.signer, gunnar_proto::LogFetchRequest { from, to, max_bytes, from_offset: 0 });
let quiet = || anyhow!("{STALLED} the primary sent nothing of Mirror.LogFetch {from}..={to} for {} s on this connection", FETCH_IDLE_WITHIN.as_secs());
self.rt.block_on(async {
let mut stream = tokio::time::timeout(FETCH_IDLE_WITHIN, client.log_fetch(req)).await.map_err(|_| quiet())?.map_err(say)?.into_inner();
feed_log_stream(&mut stream, Some((FETCH_IDLE_WITHIN, &quiet)), &|status| say(status), into).await
})
}
fn snapshot(&self, into: &mut snapshot::Stager) -> anyhow::Result<bool> {
let mut client = self.client.lock().expect("twin client").clone();
let req = gunnar_grpc::client_signer::authed(&self.signer, gunnar_proto::SnapshotFetchRequest::default());
let quiet = || anyhow!("{STALLED} the primary sent nothing of Mirror.SnapshotFetch for {} s on this connection", SNAPSHOT_IDLE_WITHIN.as_secs());
self.rt.block_on(async {
let mut stream = tokio::time::timeout(SNAPSHOT_IDLE_WITHIN, client.snapshot_fetch(req)).await.map_err(|_| quiet())?.map_err(say)?.into_inner();
while let Some(chunk) = tokio::time::timeout(SNAPSHOT_IDLE_WITHIN, stream.message()).await.map_err(|_| quiet())?.map_err(say)? {
anyhow::ensure!(chunk.seq == 0, "the primary's snapshot stream carried seq {}", chunk.seq);
let kind = u16::try_from(chunk.kind).map_err(|_| anyhow!("snapshot piece of kind {}", chunk.kind))?;
into.feed(snapshot::Piece { kind, meta: &chunk.meta, blob_len: chunk.blob_len, blob: &chunk.blob, last: chunk.last })?;
}
Ok::<_, anyhow::Error>(())
})?;
anyhow::ensure!(into.terminated().is_some(), "the primary's snapshot stream carried no SnapshotAt terminator");
Ok(true)
}
}
fn nap(poll: Duration, stop: &AtomicBool) {
let until = std::time::Instant::now() + poll;
while !stop.load(Ordering::Relaxed) {
let left = until.saturating_duration_since(std::time::Instant::now());
if left.is_zero() {
return;
}
std::thread::sleep(left.min(NAP_SLICE));
}
}
pub const NAP_SLICE: Duration = Duration::from_millis(100);
#[derive(Debug, Clone, Copy)]
pub struct AwaitingCertificate;
impl std::fmt::Display for AwaitingCertificate {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("no client certificate is configured on this box yet")
}
}
impl std::error::Error for AwaitingCertificate {}
pub const CERTIFY_GRACE: Duration = Duration::from_secs(180);
pub(crate) fn awaiting_certify(err: &anyhow::Error, since_start: Duration) -> bool {
since_start < CERTIFY_GRACE && err.downcast_ref::<AwaitingCertificate>().is_some()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Ended {
Stopped,
SnapshotNeeded,
Diverged,
Redial,
}
pub const GROWING_WAIT_POLLS: u32 = 5;
static FOLLOWING: AtomicBool = AtomicBool::new(false);
static STOP_FOLLOWING: AtomicBool = AtomicBool::new(false);
static REPOINTED: AtomicBool = AtomicBool::new(false);
pub const FOLLOWER_RETRY: Duration = Duration::from_secs(5);
pub const QUICK_REDIAL_EVERY: Duration = Duration::from_secs(30);
pub fn start_follower(stores: Arc<ServerStores>) {
start_follower_every(stores, FOLLOWER_RETRY);
}
pub const TWIN_FOLLOWER: &str = "gunnar-twin";
pub fn start_follower_every(stores: Arc<ServerStores>, retry: Duration) {
if FOLLOWING.swap(true, Ordering::SeqCst) {
return;
}
STOP_FOLLOWING.store(false, Ordering::SeqCst);
let say: nordisk_pairing::duty::Say = Arc::new(|l: &str| tracing::error!(target: gunnar_core::observe::TARGET_TWIN, "{l}"));
let started = nordisk_pairing::duty::start_with(TWIN_FOLLOWER, retry, say, move || {
let failures: Arc<Mutex<Option<Failing>>> = Arc::new(Mutex::new(None));
let mut attempt: u64 = 0;
let started = std::time::Instant::now();
let mut quick: Option<std::time::Instant> = None;
move || {
let run = || -> anyhow::Result<Ended> {
let cfg = TwinConfig::load(stores.root())?.ok_or_else(|| anyhow!("no {TWIN_CONFIG}: not a twin"))?;
let key = twin_private_key(stores.root())?;
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, primary = %cfg.primary, "following the primary");
let reporting = Arc::clone(&stores);
let source = GrpcSource::connect(&cfg.primary, key, stores.root())?.with_reporter(move || measure_report(&reporting));
Ok(Follower::new(Arc::clone(&stores), Box::new(source)).sharing_failures(Arc::clone(&failures)).run(Duration::from_secs(2), &STOP_FOLLOWING))
};
if STOP_FOLLOWING.load(Ordering::SeqCst) {
if !REPOINTED.swap(false, Ordering::SeqCst) {
FOLLOWING.store(false, Ordering::SeqCst);
return nordisk_pairing::duty::Round::Done;
}
STOP_FOLLOWING.store(false, Ordering::SeqCst);
attempt = 0;
return nordisk_pairing::duty::Round::After(Duration::ZERO);
}
attempt += 1;
match run() {
Err(err) if awaiting_certify(&err, started.elapsed()) => {
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, attempt, "the twin waits for its client certificate from the pairing (`twin certify`); dialling again");
stalled(&stores, attempt, &format!("{err:#}"), retry);
}
Err(err) => {
tracing::error!(target: gunnar_core::observe::TARGET_TWIN, %err, attempt, "the twin cannot follow; starting it again");
stalled(&stores, attempt, &format!("{err:#}"), retry);
}
Ok(Ended::Redial) => {
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, attempt, "the twin's poll loop handed back to re-dial its primary; following again");
attempt = 0;
if quick.is_none_or(|t| t.elapsed() >= QUICK_REDIAL_EVERY) {
quick = Some(std::time::Instant::now());
return nordisk_pairing::duty::Round::After(Duration::ZERO);
}
}
Ok(_) => {
if STOP_FOLLOWING.load(Ordering::SeqCst) {
return nordisk_pairing::duty::Round::After(Duration::ZERO);
}
if status(stores.root()).map(|s| s.state) == Some(State::Diverged) {
tracing::error!(target: gunnar_core::observe::TARGET_TWIN, why = %status(stores.root()).map(|s| s.why).unwrap_or_default(), "the twin stopped: the box it is pointed at is a different log");
FOLLOWING.store(false, Ordering::SeqCst);
return nordisk_pairing::duty::Round::Done;
}
if status(stores.root()).map(|s| s.state) == Some(State::SnapshotNeeded) {
tracing::error!(target: gunnar_core::observe::TARGET_TWIN, "the twin stopped at a hole its primary no longer retains: it needs `gunnar control twin catchup`");
FOLLOWING.store(false, Ordering::SeqCst);
return nordisk_pairing::duty::Round::Done;
}
tracing::error!(target: gunnar_core::observe::TARGET_TWIN, attempt, "the twin's poll loop returned without being asked to stop; starting it again");
stalled(&stores, attempt, "the poll loop returned without being asked to stop", retry);
}
}
nordisk_pairing::duty::Round::After(retry)
}
});
if let Err(e) = started {
FOLLOWING.store(false, Ordering::SeqCst);
tracing::error!(target: gunnar_core::observe::TARGET_TWIN, error = %e, "ALERT {TWIN_FOLLOWER} the follower could not start: this twin does NOT follow its primary");
}
}
pub fn onto_tunnel(stores: &ServerStores, utility: &str, overlay: &str) -> anyhow::Result<bool> {
let root = stores.root();
let (utility, overlay) = (utility.trim(), overlay.trim());
anyhow::ensure!(!overlay.is_empty(), "the primary's overlay address is required");
let Some(mut cfg) = TwinConfig::load(root)? else { return Ok(false) };
if utility.is_empty() || cfg.primary != utility || utility == overlay {
return Ok(false);
}
cfg.primary = overlay.to_string();
cfg.save(root)?;
tracing::info!(target: gunnar_core::observe::TARGET_TWIN, was = %utility, now = %overlay, "this twin follows the SAME primary over the tunnel now: joined on its utility address before WireGuard-only, its follower dials the overlay address from its next attempt");
if FOLLOWING.load(Ordering::SeqCst) {
REPOINTED.store(true, Ordering::SeqCst);
STOP_FOLLOWING.store(true, Ordering::SeqCst);
}
Ok(true)
}
pub fn repoint(stores: &ServerStores, primary: &str) -> anyhow::Result<bool> {
let root = stores.root();
let primary = primary.trim();
anyhow::ensure!(!primary.is_empty(), "the primary's address is required");
let mut cfg = TwinConfig::load(root)?.ok_or_else(|| anyhow!("no {TWIN_CONFIG}: this box is not a twin, there is nothing to re-point"))?;
if cfg.primary == primary {
return Ok(false);
}
let was = std::mem::replace(&mut cfg.primary, primary.to_string());
cfg.save(root)?;
tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, was = %was, now = %primary, "RE-POINTED: this twin's primary was recharged at another address; the follower dials it from its next attempt");
if FOLLOWING.load(Ordering::SeqCst) {
REPOINTED.store(true, Ordering::SeqCst);
STOP_FOLLOWING.store(true, Ordering::SeqCst);
}
Ok(true)
}
pub fn stop_follower() {
STOP_FOLLOWING.store(true, Ordering::SeqCst);
}
pub const FOLLOWER_NOT_RUNNING: &str = "the follower is NOT running";
fn stalled(stores: &ServerStores, attempt: u64, why: &str, retry: Duration) {
if maintenance::is_held(stores.root()) {
return;
}
let last = status(stores.root());
set_status(stores.root(), TwinStatus {
applied_seq: stores.txlog().head_seq(),
primary_head_seq: last.as_ref().map(|s| s.primary_head_seq).unwrap_or(0),
state: State::Unreachable,
why: format!("{FOLLOWER_NOT_RUNNING} (attempt {attempt}, starting again in {} s): {why}", retry.as_secs_f32()),
last_answer_unix_ms: last.as_ref().map(|s| s.last_answer_unix_ms).unwrap_or(0),
last_level_unix_ms: 0,
});
}
pub fn restore(from: Arc<Log>, into: Arc<ServerStores>, source_log_id: &str) -> anyhow::Result<u64> {
restore_from(Box::new(from), into, source_log_id)
}
pub fn restore_from(source: Box<dyn LogSource>, into: Arc<ServerStores>, source_log_id: &str) -> anyhow::Result<u64> {
restore_begin(&into)?;
let f = Follower::new(Arc::clone(&into), source);
loop {
match f.step()? {
Progress::Applied { .. } | Progress::SnapshotInstalled { .. } => continue,
Progress::CaughtUp { .. } => return restore_finish(&into, 0, source_log_id),
Progress::SnapshotNeeded { wanted, first } => {
return Err(anyhow!("the source starts at {first}; a restore needs {wanted}"))
}
Progress::Diverged { own, primary_head } => {
return Err(anyhow!(
"the source's head is {primary_head}, below the {own} already on this box — it is not this box's log"
))
}
}
}
}
pub mod snapshot;
pub mod intake;
pub struct LocalSource {
pub stores: Arc<ServerStores>,
pub retained_from: u64,
}
impl LocalSource {
fn whole(&self) -> anyhow::Result<()> {
match self.stores.is_twin().then(|| snapshot::not_whole(self.stores.root())).flatten() {
Some(why) => anyhow::bail!("{why}. Nothing is read from it"),
None => Ok(()),
}
}
}
impl LogSource for LocalSource {
fn head(&self) -> anyhow::Result<(u64, u64)> {
let log = self.stores.txlog();
Ok((log.first_seq().max(self.retained_from), log.head_seq()))
}
fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
self.whole()?;
let first = self.head()?.0;
if from < first {
return Err(txlog::Error::NotRetained { seq: from, first }.into());
}
Ok(fetch_range(self.stores.txlog(), from, to, max_bytes)?)
}
fn fetch_into(&self, from: u64, to: u64, max_bytes: u64, into: &mut dyn FrameSink) -> anyhow::Result<()> {
self.whole()?;
let first = self.head()?.0;
if from < first {
return Err(txlog::Error::NotRetained { seq: from, first }.into());
}
stream_range(self.stores.txlog(), from, to, max_bytes, into).map(drop)
}
fn snapshot(&self, into: &mut snapshot::Stager) -> anyhow::Result<bool> {
snapshot::serve(&self.stores, &mut |p| into.feed(p))?;
Ok(true)
}
}
#[cfg(test)]
mod tests;
pub fn say_if_stores_were_applied(stores: &ServerStores) {
let repos = stores.root().join(crate::stores::REPOS_DIR);
let stale = std::fs::read_dir(&repos).map(|d| d.flatten().count()).unwrap_or(0);
if stale > 0 {
tracing::warn!(
target: gunnar_core::observe::TARGET_TWIN,
path = %repos.display(),
namespaces = stale,
"this twin holds repository stores an OLDER build applied; a twin only stores (2026-09-22) and nothing reads or updates them — \
they are stale bytes on the twin's volume, kept rather than deleted by an open"
);
}
}
#[cfg(test)]
mod skew_tests {
#[test]
fn a_poll_refused_for_skew_is_told_from_any_other_failure() {
let skew = anyhow::anyhow!("status: Unauthenticated, message: \"the credential is -7198470 ms away from this server's clock; the window is ±300000 ms\"");
assert!(super::refused_for_skew(&skew));
let stale = gunnar_grpc::auth::AuthFailure::Stale { skew_ms: -7_198_470 }.status();
assert!(super::refused_for_skew(&anyhow::anyhow!("{stale}")), "the gate's own Stale status: {stale}");
for other in ["status: Unauthenticated, message: \"this credential has already been used\"", "transport error: connection refused", "the primary is not answering"] {
assert!(!super::refused_for_skew(&anyhow::anyhow!("{other}")), "{other}");
}
}
}
#[cfg(test)]
mod own_load_tests {
#[test]
fn a_twin_reports_its_own_cores_and_load() {
let (busy, total) = super::own_load();
assert!(total.is_some_and(|t| t >= 1_000 && t % 1_000 == 0), "{total:?}");
assert!(busy.is_some(), "/proc/loadavg is readable here");
}
}
#[cfg(test)]
mod awaiting_certificate_tests {
use super::*;
#[test]
fn a_missing_certificate_is_awaited_only_inside_the_grace_window() {
let err = anyhow::Error::new(AwaitingCertificate).context("this twin\x27s primary is 10.44.0.2:50051, which is another machine");
assert!(awaiting_certify(&err, Duration::from_secs(3)));
assert!(!awaiting_certify(&err, CERTIFY_GRACE));
assert!(format!("{err:#}").contains("no client certificate is configured on this box yet"));
}
#[test]
fn any_other_failure_stays_an_error() {
let err = anyhow::anyhow!("connection refused");
assert!(!awaiting_certify(&err, Duration::from_secs(1)));
}
#[test]
fn a_remote_primary_with_no_certificate_refuses_as_awaiting() {
let dir = tempfile::tempdir().unwrap();
let key = gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap()).unwrap();
let err = GrpcSource::connect("10.44.0.2:50051", key, dir.path()).err().expect("refused");
assert!(err.downcast_ref::<AwaitingCertificate>().is_some(), "{err:#}");
assert!(format!("{err:#}").contains("It will not"), "{err:#}");
}
}