v0.1.3: make the vault survive a power cut, and recover if it doesn't
The vault was being written with `fs::write(tmp)` → `rename(tmp, path)`. The
rename is atomic, so readers never saw a half-written file — but atomicity is
not durability. Nothing fsynced the new bytes before the rename, so on ext4
(`data=ordered`, the default) a crash or hard power-off could commit the
rename while the data it published never landed. The result is precisely the
failure that stopped the container booting: `vault.json` present, zero bytes,
"EOF while parsing a value at line 1 column 0".
Two things made it worse. The write happened on *every* push tick, so a
ten-account vault was rewriting all of its secrets thousands of times a day,
and there was no backup to fall back on — a damaged file meant a crash loop
with the web UI locked behind the service that would not start.
Storage
* `store_io::write_atomic` fsyncs the new bytes, renames, then fsyncs the
directory. That ordering is the whole fix; the module docs explain why.
* Every vault write refreshes `vault.json.bak`, taken *after* the new file is
safely in place so the two always agree. Backing up first would leave the
backup a full change behind and quietly lose the most recent edit.
* `Vault::load` falls back to the backup when `vault.json` is unparseable,
moves the damaged file to `vault.json.corrupt-<ts>` rather than deleting
it, and writes the recovered accounts back so the next boot is ordinary.
With no usable backup it still refuses to start, as before.
Write volume
* Discord bookkeeping — message ids, last codes, last errors — moves to its
own `push-state.json`, batched and flushed at most every five seconds, with
a final flush on shutdown. Codes roll constantly; accounts change only when
the operator edits them, and keeping the two together was what put the
secrets in the path of every rotation.
* `push-state.json` is a cache by construction: losing or corrupting it costs
a reposted tile, never a secret, so it is never a reason to refuse to boot.
Tests: recovery from a zero-length vault, backup retention, write-back on
recovery, refusal when both files are bad, reload of tile ids, batching, and
that a rotation leaves `vault.json` byte-identical. 143 passing.
This commit is contained in:
1 parent
7751ebe9b3
commit
b21ade05eb
8 files changed
+813
-36
No files matched your search
Generated
+6
-6
@@ -436,7 +436,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "da-core"
|
||||
version = "0.1.2"
|
||||
version = "0.1.3"
|
||||
dependencies = [
|
||||
"argon2",
|
||||
"base32",
|
||||
@@ -461,7 +461,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "da-discord"
|
||||
version = "0.1.2"
|
||||
version = "0.1.3"
|
||||
dependencies = [
|
||||
"axum",
|
||||
"da-core",
|
||||
@@ -478,7 +478,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "da-http-api"
|
||||
version = "0.1.2"
|
||||
version = "0.1.3"
|
||||
dependencies = [
|
||||
"axum",
|
||||
"base64 0.22.1",
|
||||
@@ -500,7 +500,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "da-render"
|
||||
version = "0.1.2"
|
||||
version = "0.1.3"
|
||||
dependencies = [
|
||||
"image",
|
||||
"resvg",
|
||||
@@ -512,7 +512,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "da-webui"
|
||||
version = "0.1.2"
|
||||
version = "0.1.3"
|
||||
|
||||
[[package]]
|
||||
name = "data-url"
|
||||
@@ -542,7 +542,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "discord-authenticator"
|
||||
version = "0.1.2"
|
||||
version = "0.1.3"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"axum",
|
||||
|
||||
+1
-1
@@ -10,7 +10,7 @@ members = [
|
||||
]
|
||||
|
||||
[workspace.package]
|
||||
version = "0.1.2"
|
||||
version = "0.1.3"
|
||||
edition = "2021"
|
||||
rust-version = "1.90"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -28,6 +28,7 @@ pull), no database, no external services beyond Discord itself.
|
||||
- [Using it](#using-it)
|
||||
- [Configuration](#configuration)
|
||||
- [Architecture](#architecture)
|
||||
- [Troubleshooting](#troubleshooting)
|
||||
- [Development](#development)
|
||||
|
||||
---
|
||||
@@ -113,6 +114,12 @@ What the app does hold up its end on:
|
||||
|
||||
- TOTP secrets are encrypted at rest with **XChaCha20-Poly1305**; `vault.json`
|
||||
contains ciphertext only, and the metadata beside it is deliberately readable.
|
||||
- `vault.json` is written atomically *and* durably — new bytes are fsynced
|
||||
before the rename that publishes them, and the directory is fsynced after —
|
||||
so an unclean shutdown can't leave a truncated file behind. The previous
|
||||
contents are kept in `vault.json.bak`, and startup falls back to it if
|
||||
`vault.json` is ever unreadable, moving the damaged file aside as
|
||||
`vault.json.corrupt-<timestamp>` rather than refusing to boot.
|
||||
- The admin password is hashed with **Argon2id**.
|
||||
- The vault key lives in `vault.key` (mode `0600`), or set `DA_MASTER_KEY` to
|
||||
keep it off the data volume entirely — then a leaked copy of that volume is
|
||||
@@ -154,6 +161,21 @@ Bridge networking is correct — this makes outbound HTTPS calls and serves one
|
||||
port. Map `/var/lib/discord-authenticator` to somewhere on `appdata` and **back
|
||||
that path up**: losing it loses every stored secret.
|
||||
|
||||
### What lives in the data directory
|
||||
|
||||
| File | Holds | If it's lost |
|
||||
| --- | --- | --- |
|
||||
| `vault.json` | Every account, with its secret sealed | Everything — this is the file to back up |
|
||||
| `vault.json.bak` | The same accounts, as of the last change | Nothing, unless `vault.json` is damaged |
|
||||
| `vault.key` | The vault key (unless `DA_MASTER_KEY` is set) | Every secret becomes unreadable |
|
||||
| `push-state.json` | Discord message ids and last-known push errors | Nothing — tiles are reposted on the next tick |
|
||||
| `auth.json`, `settings.json` | Admin login and webhook config | You redo first-run setup |
|
||||
|
||||
`push-state.json` is deliberately separate from the vault. Codes roll every 30
|
||||
seconds, so the bookkeeping changes constantly while the accounts themselves
|
||||
change only when you edit them; keeping them together meant every rotation
|
||||
rewrote all the secrets.
|
||||
|
||||
### Building and publishing the image
|
||||
|
||||
```bash
|
||||
@@ -285,7 +307,7 @@ JetBrains Mono. Light and dark both ship.
|
||||
|
||||
### Testing
|
||||
|
||||
126 tests, including:
|
||||
143 tests, including:
|
||||
|
||||
- The **RFC 6238 test vectors** for SHA-1, SHA-256 and SHA-512.
|
||||
- Real **QR round-trips** — clean, 1-pixel-per-module, oversized, rotated, JPEG,
|
||||
@@ -300,6 +322,11 @@ JetBrains Mono. Light and dark both ship.
|
||||
again, that a deleted tile is reposted, that a dead webhook is surfaced
|
||||
instead of hammered, that concurrent rollovers are paced apart, and that a
|
||||
rename refreshes the tile immediately without re-posting it or looping.
|
||||
- **Crash recovery**: that a zero-length `vault.json` is recovered from the
|
||||
backup, that the damaged file is kept rather than deleted, that the recovered
|
||||
accounts are written back so the next boot is ordinary, that an unusable
|
||||
backup still refuses to boot rather than starting empty, and that a code
|
||||
rotation does not rewrite the file holding the secrets.
|
||||
|
||||
```bash
|
||||
cargo test --workspace
|
||||
@@ -308,6 +335,39 @@ cargo clippy --workspace --all-targets
|
||||
|
||||
---
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### The container won't start: "could not load the vault"
|
||||
|
||||
The app refuses to start on an unreadable `vault.json` because the alternative
|
||||
— booting with an empty vault — invites you to re-add accounts you already have.
|
||||
|
||||
Since v0.1.3 it recovers on its own first: if `vault.json.bak` parses, the
|
||||
accounts are loaded from it, the damaged file is moved to
|
||||
`vault.json.corrupt-<timestamp>`, and a fresh `vault.json` is written. You will
|
||||
see one `ERROR` line saying so, then normal startup.
|
||||
|
||||
If it still refuses, both files are unreadable and there is nothing to recover
|
||||
from automatically. Get the file out of the way and restart:
|
||||
|
||||
```bash
|
||||
cd /mnt/user/appdata/discord-authenticator
|
||||
mv vault.json "vault.json.broken-$(date +%s)"
|
||||
docker restart Discord-Authenticator
|
||||
```
|
||||
|
||||
The service will start with an empty vault, so re-import your accounts from
|
||||
your own backups (`otpauth://` links or a Google Authenticator export). Keep
|
||||
`vault.key` — without it, nothing in the old file can be decrypted even if the
|
||||
bytes are recovered later.
|
||||
|
||||
**If you hit this, upgrading matters more than the workaround.** The cause was
|
||||
a rename being journalled ahead of the data it published, which is fixed by
|
||||
fsyncing before the rename; see `crates/core/src/store_io.rs` for the ordering
|
||||
and why it is the whole trick.
|
||||
|
||||
---
|
||||
|
||||
## Development
|
||||
|
||||
```bash
|
||||
|
||||
@@ -12,6 +12,7 @@ pub mod crypto;
|
||||
pub mod error;
|
||||
pub mod migration;
|
||||
pub mod settings;
|
||||
pub mod store_io;
|
||||
pub mod totp;
|
||||
pub mod vault;
|
||||
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
//! Crash-safe JSON persistence.
|
||||
//!
|
||||
//! Every store in this crate used to do its own
|
||||
//! `fs::write(tmp)` → `fs::rename(tmp, path)`. The rename is atomic, so a
|
||||
//! reader never sees a half-written file — but *durability* is a separate
|
||||
//! promise, and that pattern does not make it.
|
||||
//!
|
||||
//! Without an `fsync` before the rename, the new file's data blocks are still
|
||||
//! only in the page cache when the rename is journalled. On ext4 (`data=ordered`,
|
||||
//! the default) a crash or hard power-off can therefore commit the rename while
|
||||
//! the data never lands, leaving the destination **present and zero-length**.
|
||||
//! That is exactly the failure this module exists to prevent: a `vault.json`
|
||||
//! that exists, parses as nothing, and refuses to let the service start.
|
||||
//!
|
||||
//! To make the rename durable the *data* must be on disk before it happens, and
|
||||
//! the rename itself must then be flushed by syncing the containing directory.
|
||||
//! That ordering is the whole trick:
|
||||
//!
|
||||
//! 1. write the new bytes to a temp file in the same directory
|
||||
//! 2. `fsync` the temp file — now its contents survive a power cut
|
||||
//! 3. `rename` it over the destination — atomic *and* durable
|
||||
//! 4. `fsync` the directory — persists the rename itself
|
||||
//!
|
||||
//! The temp file lives in the destination's own directory so the rename stays
|
||||
//! within one filesystem; across filesystems `rename` degrades to copy-then-
|
||||
//! delete and is no longer atomic.
|
||||
|
||||
use std::fs::{self, File};
|
||||
use std::io::Write;
|
||||
use std::path::Path;
|
||||
|
||||
/// Write `bytes` to `path` atomically and durably.
|
||||
///
|
||||
/// See the module docs for the ordering this relies on. `path` may name a file
|
||||
/// that does not exist yet, but its parent directory must.
|
||||
///
|
||||
/// The temp name is derived from the destination, so concurrent calls for the
|
||||
/// same destination must be serialized by the caller — every store here holds
|
||||
/// a lock across its write for that reason. Nothing stops a *second process*
|
||||
/// from being pointed at the same data directory; one writer per directory
|
||||
/// remains the assumption.
|
||||
pub fn write_atomic(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
|
||||
let dir = path.parent().unwrap_or_else(|| Path::new("."));
|
||||
let tmp = temp_path(path);
|
||||
|
||||
// Scope the file handle so it is closed before the rename. Renaming an
|
||||
// open file is fine on Unix, but closing first keeps the ordering obvious
|
||||
// and behaves identically on platforms where it is not.
|
||||
write_and_sync(&tmp, bytes)?;
|
||||
|
||||
if let Err(e) = fs::rename(&tmp, path) {
|
||||
let _ = fs::remove_file(&tmp);
|
||||
return Err(e);
|
||||
}
|
||||
sync_dir(dir);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// `vault.json` → `vault.json.tmp`.
|
||||
///
|
||||
/// `with_extension` rather than `with_file_name` so a path that has no
|
||||
/// extension still gains one instead of being replaced outright.
|
||||
fn temp_path(path: &Path) -> std::path::PathBuf {
|
||||
path.with_extension("json.tmp")
|
||||
}
|
||||
|
||||
/// Create the temp file, write the bytes, and flush them all the way to the
|
||||
/// device.
|
||||
///
|
||||
/// The mode is set at creation rather than chmod-ed afterwards, so the bytes
|
||||
/// are never briefly readable by anyone the umask would have allowed.
|
||||
fn write_and_sync(tmp: &Path, bytes: &[u8]) -> std::io::Result<()> {
|
||||
let mut opts = fs::OpenOptions::new();
|
||||
opts.write(true).create(true).truncate(true);
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::OpenOptionsExt;
|
||||
opts.mode(0o600);
|
||||
}
|
||||
let mut f = opts.open(tmp)?;
|
||||
f.write_all(bytes)?;
|
||||
// The point of the whole exercise: without this the rename below can be
|
||||
// journalled ahead of the data it is supposed to be publishing.
|
||||
f.sync_all()?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Persist a directory entry change. Best effort — a filesystem that refuses
|
||||
/// `fsync` on a directory does not make the write above any less atomic, so
|
||||
/// this must never turn a successful save into an error.
|
||||
fn sync_dir(dir: &Path) {
|
||||
#[cfg(unix)]
|
||||
if let Ok(d) = File::open(dir) {
|
||||
let _ = d.sync_all();
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
let _ = dir;
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use tempfile::tempdir;
|
||||
|
||||
#[test]
|
||||
fn writes_and_replaces_a_file() {
|
||||
let dir = tempdir().unwrap();
|
||||
let path = dir.path().join("thing.json");
|
||||
|
||||
write_atomic(&path, b"first").unwrap();
|
||||
assert_eq!(fs::read(&path).unwrap(), b"first");
|
||||
|
||||
// Replacing an existing file must leave the new bytes, not append and
|
||||
// not leave the old length.
|
||||
write_atomic(&path, b"second").unwrap();
|
||||
assert_eq!(fs::read(&path).unwrap(), b"second");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn creates_a_file_that_does_not_exist_yet() {
|
||||
let dir = tempdir().unwrap();
|
||||
let path = dir.path().join("fresh.json");
|
||||
assert!(!path.exists());
|
||||
write_atomic(&path, b"{}").unwrap();
|
||||
assert_eq!(fs::read(&path).unwrap(), b"{}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn leaves_no_temp_file_behind() {
|
||||
let dir = tempdir().unwrap();
|
||||
let path = dir.path().join("thing.json");
|
||||
write_atomic(&path, b"body").unwrap();
|
||||
|
||||
let leftovers: Vec<_> = fs::read_dir(dir.path())
|
||||
.unwrap()
|
||||
.flatten()
|
||||
.map(|e| e.file_name().to_string_lossy().into_owned())
|
||||
.filter(|n| n.contains(".tmp"))
|
||||
.collect();
|
||||
assert!(leftovers.is_empty(), "stray temp files: {leftovers:?}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reports_an_error_rather_than_panicking_when_the_directory_is_missing() {
|
||||
let dir = tempdir().unwrap();
|
||||
let path = dir.path().join("nope").join("thing.json");
|
||||
assert!(write_atomic(&path, b"body").is_err());
|
||||
}
|
||||
}
|
||||
+572
-28
@@ -11,16 +11,50 @@
|
||||
//! Everywhere else falling back to defaults is the friendly move; here it
|
||||
//! would silently present an empty vault and invite the operator to re-add
|
||||
//! accounts they already have, so we refuse to start instead.
|
||||
//!
|
||||
//! ## Why there are two files
|
||||
//!
|
||||
//! `vault.json` holds the accounts — the irreplaceable part. `push-state.json`
|
||||
//! holds the Discord bookkeeping: message ids, last-sent codes, last error.
|
||||
//! They are split because they change at wildly different rates. One code
|
||||
//! rotation is one push per account, so at a 5-second tick a ten-account vault
|
||||
//! rewrites its bookkeeping thousands of times a day, while the accounts
|
||||
//! themselves change only when the operator adds, renames or removes one.
|
||||
//!
|
||||
//! Keeping them in one file meant every one of those ticks rewrote 100% of the
|
||||
//! secrets. Two files mean the secret-bearing file is touched only on real
|
||||
//! edits, and the file that *is* written constantly is a pure cache: losing it
|
||||
//! costs a duplicate Discord tile, never a TOTP secret.
|
||||
//!
|
||||
//! ## Surviving a power cut
|
||||
//!
|
||||
//! Both files go through [`crate::store_io::write_atomic`], which fsyncs the
|
||||
//! new bytes before renaming them into place and then fsyncs the directory.
|
||||
//! That ordering matters more than the atomicity: a rename can reach the
|
||||
//! journal before the data it publishes, which on ext4 is exactly how you end
|
||||
//! up with a file that exists and is zero bytes long. `vault.json.bak` keeps
|
||||
//! the previous good copy, and [`Vault::load`] falls back to it rather than
|
||||
//! leaving the operator with a service that will not boot.
|
||||
|
||||
use crate::crypto::{Sealed, VaultKey};
|
||||
use crate::store_io::write_atomic;
|
||||
use crate::totp::{Algorithm, OtpAuth, Totp};
|
||||
use crate::{Error, Result};
|
||||
use parking_lot::RwLock;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashMap;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use time::OffsetDateTime;
|
||||
|
||||
/// How long a Discord bookkeeping change may sit unflushed in memory.
|
||||
///
|
||||
/// The push loop calls [`Vault::flush_state`] on every pass, and a pass
|
||||
/// happens every few seconds; this bounds the write rate independently of how
|
||||
/// many accounts are pushing. Nothing here is worth a synchronous disk write,
|
||||
/// because losing it only costs one re-created Discord message.
|
||||
const STATE_FLUSH_AFTER: std::time::Duration = std::time::Duration::from_secs(5);
|
||||
|
||||
/// Live-message bookkeeping for one account's Discord tile.
|
||||
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
|
||||
#[serde(default)]
|
||||
@@ -149,11 +183,51 @@ struct VaultFile {
|
||||
accounts: Vec<Account>,
|
||||
}
|
||||
|
||||
/// The Discord bookkeeping for every account, kept apart from the secrets.
|
||||
///
|
||||
/// Keyed by account id so the vault file never has to be rewritten when a code
|
||||
/// rotates, and so an entry for an account that has since been removed is
|
||||
/// simply ignored on load rather than needing a cleanup pass.
|
||||
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
|
||||
struct PushStateFile {
|
||||
#[serde(default)]
|
||||
accounts: HashMap<String, DiscordState>,
|
||||
}
|
||||
|
||||
/// Bookkeeping changes that have not reached disk yet.
|
||||
#[derive(Debug, Default)]
|
||||
struct PendingState {
|
||||
dirty: bool,
|
||||
last_flush: Option<std::time::Instant>,
|
||||
}
|
||||
|
||||
impl PendingState {
|
||||
/// True when this change is due to be written: something is dirty *and*
|
||||
/// the last write is old enough. The first change after a quiet period
|
||||
/// flushes immediately, so a fresh message id is on disk within the tick
|
||||
/// that created it rather than waiting out a timer.
|
||||
fn due(&self) -> bool {
|
||||
self.dirty
|
||||
&& self
|
||||
.last_flush
|
||||
.is_none_or(|t| t.elapsed() >= STATE_FLUSH_AFTER)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct Vault {
|
||||
path: Arc<PathBuf>,
|
||||
/// Where the previous good `vault.json` is kept, so a file damaged by a
|
||||
/// crash can be recovered instead of ending the service's life.
|
||||
backup_path: Arc<PathBuf>,
|
||||
state_path: Arc<PathBuf>,
|
||||
key: Arc<VaultKey>,
|
||||
inner: Arc<RwLock<Vec<Account>>>,
|
||||
/// Serializes whole-file writes. Two threads persisting at once would
|
||||
/// otherwise interleave on the same temp path and could rename a partial
|
||||
/// file into place.
|
||||
write_lock: Arc<parking_lot::Mutex<()>>,
|
||||
pending: Arc<parking_lot::Mutex<PendingState>>,
|
||||
}
|
||||
|
||||
#[allow(clippy::missing_fields_in_debug)] // secrets must never reach a log line
|
||||
@@ -169,30 +243,87 @@ impl Vault {
|
||||
/// Load `<work_dir>/vault.json`, or start empty if it doesn't exist yet.
|
||||
///
|
||||
/// A file that exists but won't parse is an error, not a fallback — see
|
||||
/// the module docs.
|
||||
/// the module docs. Before giving up we try `vault.json.bak`, the previous
|
||||
/// good copy, because refusing to boot is only the right answer when there
|
||||
/// is genuinely nothing left to load: the operator is otherwise locked out
|
||||
/// of every account they hold, with the web UI that could fix it being the
|
||||
/// very thing that won't start.
|
||||
pub fn load(work_dir: &Path, key: VaultKey) -> Result<Self> {
|
||||
let path = work_dir.join("vault.json");
|
||||
let accounts = match std::fs::read_to_string(&path) {
|
||||
Ok(text) => {
|
||||
serde_json::from_str::<VaultFile>(&text)
|
||||
.map_err(|e| {
|
||||
Error::invalid(format!(
|
||||
"{} exists but could not be parsed ({e}). Refusing to start \
|
||||
with an empty vault — move the file aside to start fresh.",
|
||||
path.display()
|
||||
))
|
||||
})?
|
||||
.accounts
|
||||
}
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Vec::new(),
|
||||
Err(e) => return Err(e.into()),
|
||||
};
|
||||
let backup_path = backup_of(&path);
|
||||
let (accounts, recovered) = load_accounts(&path, &backup_path)?;
|
||||
tracing::info!(target: "da::vault", count = accounts.len(), "vault loaded");
|
||||
Ok(Self {
|
||||
|
||||
let state_path = work_dir.join("push-state.json");
|
||||
let v = Self {
|
||||
path: Arc::new(path),
|
||||
backup_path: Arc::new(backup_path),
|
||||
state_path: Arc::new(state_path.clone()),
|
||||
key: Arc::new(key),
|
||||
inner: Arc::new(RwLock::new(accounts)),
|
||||
})
|
||||
write_lock: Arc::new(parking_lot::Mutex::new(())),
|
||||
pending: Arc::new(parking_lot::Mutex::new(PendingState::default())),
|
||||
};
|
||||
v.restore_push_state(&state_path);
|
||||
if recovered {
|
||||
// Put the recovered accounts back where a vault belongs. Until
|
||||
// this happens the only copy on disk is the `.bak`, so a second
|
||||
// unclean stop would leave nothing to recover from; doing it now
|
||||
// also means the next boot takes the ordinary path.
|
||||
v.persist();
|
||||
}
|
||||
Ok(v)
|
||||
}
|
||||
|
||||
/// Re-attach the Discord bookkeeping to the loaded accounts.
|
||||
///
|
||||
/// Entirely best effort: this file is a cache, so a missing, unreadable or
|
||||
/// stale one costs a re-posted tile and nothing more. Deliberately *not* a
|
||||
/// reason to refuse to start, which is the whole point of keeping it
|
||||
/// separate from the vault.
|
||||
fn restore_push_state(&self, state_path: &Path) {
|
||||
let text = match std::fs::read_to_string(state_path) {
|
||||
Ok(t) => t,
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return,
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
target: "da::vault",
|
||||
path = %state_path.display(),
|
||||
"could not read push state, tiles will be re-created: {e}"
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
let parsed = match serde_json::from_str::<PushStateFile>(&text) {
|
||||
Ok(p) => p,
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
target: "da::vault",
|
||||
path = %state_path.display(),
|
||||
"push state is unreadable, tiles will be re-created: {e}"
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let mut g = self.inner.write();
|
||||
let mut restored = 0usize;
|
||||
for a in g.iter_mut() {
|
||||
// An entry with no message id carries nothing worth restoring.
|
||||
if let Some(state) = parsed.accounts.get(&a.id) {
|
||||
if state.message_id.is_some() {
|
||||
a.discord = state.clone();
|
||||
restored += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
if restored > 0 {
|
||||
tracing::info!(
|
||||
target: "da::vault",
|
||||
count = restored,
|
||||
"restored Discord tile ids"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
pub fn len(&self) -> usize {
|
||||
@@ -427,7 +558,7 @@ impl Vault {
|
||||
a.discord.last_error = None;
|
||||
a.discord.last_push_at = Some(OffsetDateTime::now_utc());
|
||||
}
|
||||
self.persist();
|
||||
self.touch_state();
|
||||
}
|
||||
|
||||
/// Record a delivery failure against one account.
|
||||
@@ -444,7 +575,7 @@ impl Vault {
|
||||
}
|
||||
a.discord.last_error = Some(error.to_string());
|
||||
}
|
||||
self.persist();
|
||||
self.touch_state();
|
||||
}
|
||||
|
||||
/// Forget an account's message id — used when Discord reports the
|
||||
@@ -458,7 +589,7 @@ impl Vault {
|
||||
a.discord.message_id = None;
|
||||
a.discord.last_code = None;
|
||||
}
|
||||
self.persist();
|
||||
self.touch_state();
|
||||
}
|
||||
|
||||
/// Drop every stored message id. Called when the webhook URL changes:
|
||||
@@ -470,7 +601,11 @@ impl Vault {
|
||||
a.discord = DiscordState::default();
|
||||
}
|
||||
}
|
||||
self.persist();
|
||||
self.touch_state();
|
||||
// A webhook change is a deliberate, rare operator action and the
|
||||
// caller is about to tell them it worked, so make it durable now
|
||||
// rather than on the next tick.
|
||||
self.flush_state_now();
|
||||
}
|
||||
|
||||
/// Is an identical account already stored?
|
||||
@@ -558,7 +693,14 @@ impl Vault {
|
||||
Ok(outcome)
|
||||
}
|
||||
|
||||
/// Write the accounts out.
|
||||
///
|
||||
/// Called only when the account list itself changes. Discord bookkeeping
|
||||
/// rides in [`Vault::state_path`] instead and never reaches this function,
|
||||
/// so the secrets are rewritten on operator edits rather than on every
|
||||
/// code rotation — see the module docs.
|
||||
fn persist(&self) {
|
||||
let _guard = self.write_lock.lock();
|
||||
let snapshot = VaultFile {
|
||||
accounts: self.inner.read().clone(),
|
||||
};
|
||||
@@ -572,15 +714,150 @@ impl Vault {
|
||||
if let Some(parent) = self.path.parent() {
|
||||
let _ = std::fs::create_dir_all(parent);
|
||||
}
|
||||
let tmp = self.path.with_extension("json.tmp");
|
||||
if let Err(e) = std::fs::write(&tmp, body) {
|
||||
tracing::error!(target: "da::vault", "could not write vault temp file: {e}");
|
||||
if let Err(e) = write_atomic(&self.path, &body) {
|
||||
tracing::error!(target: "da::vault", "could not commit vault file: {e}");
|
||||
return;
|
||||
}
|
||||
if let Err(e) = std::fs::rename(&tmp, self.path.as_path()) {
|
||||
tracing::error!(target: "da::vault", "could not commit vault file: {e}");
|
||||
// Refresh the backup *after* the new vault is safely in place, so the
|
||||
// two files always hold the same accounts. The tempting order — back
|
||||
// up the old file, then overwrite it — leaves the backup a full change
|
||||
// behind, which quietly loses the most recent edit if the vault is
|
||||
// later damaged. Recovery is only worth having if it recovers
|
||||
// everything.
|
||||
if let Err(e) = std::fs::copy(self.path.as_path(), self.backup_path.as_path()) {
|
||||
tracing::warn!(target: "da::vault", "could not refresh vault backup: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
/// Mark the Discord bookkeeping as changed.
|
||||
///
|
||||
/// Cheap on purpose: the actual write is deferred to
|
||||
/// [`Vault::flush_state`], which the push loop calls once per pass.
|
||||
fn touch_state(&self) {
|
||||
let mut g = self.pending.lock();
|
||||
g.dirty = true;
|
||||
}
|
||||
|
||||
/// Write `push-state.json` if anything has changed and enough time has
|
||||
/// passed. Safe and cheap to call on every push tick.
|
||||
pub fn flush_state(&self) {
|
||||
{
|
||||
let g = self.pending.lock();
|
||||
if !g.due() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
self.write_state();
|
||||
}
|
||||
|
||||
/// Write `push-state.json` unconditionally — for shutdown, where there is
|
||||
/// no next tick to catch a deferred write.
|
||||
pub fn flush_state_now(&self) {
|
||||
self.write_state();
|
||||
}
|
||||
|
||||
fn write_state(&self) {
|
||||
let _guard = self.write_lock.lock();
|
||||
let body = {
|
||||
let accounts = self.inner.read();
|
||||
let file = PushStateFile {
|
||||
accounts: accounts
|
||||
.iter()
|
||||
.filter(|a| a.discord.message_id.is_some() || a.discord.last_error.is_some())
|
||||
.map(|a| (a.id.clone(), a.discord.clone()))
|
||||
.collect(),
|
||||
};
|
||||
match serde_json::to_vec(&file) {
|
||||
Ok(b) => b,
|
||||
Err(e) => {
|
||||
tracing::error!(target: "da::vault", "could not serialize push state: {e}");
|
||||
return;
|
||||
}
|
||||
}
|
||||
};
|
||||
if let Some(parent) = self.state_path.parent() {
|
||||
let _ = std::fs::create_dir_all(parent);
|
||||
}
|
||||
if let Err(e) = write_atomic(&self.state_path, &body) {
|
||||
tracing::error!(target: "da::vault", "could not write push state: {e}");
|
||||
// Leave it dirty so the next tick tries again rather than
|
||||
// silently forgetting the change.
|
||||
return;
|
||||
}
|
||||
let mut g = self.pending.lock();
|
||||
g.dirty = false;
|
||||
g.last_flush = Some(std::time::Instant::now());
|
||||
}
|
||||
}
|
||||
|
||||
/// `vault.json` → `vault.json.bak`.
|
||||
fn backup_of(path: &Path) -> PathBuf {
|
||||
let mut name = path
|
||||
.file_name()
|
||||
.map_or_else(|| std::ffi::OsString::from("vault"), ToOwned::to_owned);
|
||||
name.push(".bak");
|
||||
path.with_file_name(name)
|
||||
}
|
||||
|
||||
/// Read the accounts, falling back to the backup if the main file is damaged.
|
||||
///
|
||||
/// Refusing to start on a bad vault is right — silently starting empty invites
|
||||
/// the operator to re-add accounts they already have. But there is a large gap
|
||||
/// between "won't parse" and "unrecoverable", and the backup lives in it:
|
||||
/// whenever one exists and does parse, the honest move is to use it and say so
|
||||
/// loudly, keeping the damaged original for inspection.
|
||||
fn load_accounts(path: &Path, backup_path: &Path) -> Result<(Vec<Account>, bool)> {
|
||||
let text = match std::fs::read_to_string(path) {
|
||||
Ok(text) => text,
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((Vec::new(), false)),
|
||||
Err(e) => return Err(e.into()),
|
||||
};
|
||||
|
||||
match serde_json::from_str::<VaultFile>(&text) {
|
||||
Ok(file) => Ok((file.accounts, false)),
|
||||
Err(primary_err) => match read_backup(backup_path) {
|
||||
Some(accounts) => {
|
||||
let quarantined = quarantine(path);
|
||||
tracing::error!(
|
||||
target: "da::vault",
|
||||
path = %path.display(),
|
||||
backup = %backup_path.display(),
|
||||
recovered = accounts.len(),
|
||||
quarantined = %quarantined.as_deref().map_or_else(
|
||||
|| "could not move aside".to_string(),
|
||||
|p| p.display().to_string()
|
||||
),
|
||||
"vault.json could not be parsed ({primary_err}); recovered the accounts \
|
||||
from the backup. The damaged file is kept alongside for inspection — \
|
||||
compare it against the backup if anything you added recently is missing."
|
||||
);
|
||||
Ok((accounts, true))
|
||||
}
|
||||
None => Err(Error::invalid(format!(
|
||||
"{} exists but could not be parsed ({primary_err}), and the backup at {} \
|
||||
could not be used either. Refusing to start with an empty vault — move the \
|
||||
file aside to start fresh.",
|
||||
path.display(),
|
||||
backup_path.display()
|
||||
))),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse the backup, or `None` if it is missing or not parseable as a vault.
|
||||
fn read_backup(backup_path: &Path) -> Option<Vec<Account>> {
|
||||
let text = std::fs::read_to_string(backup_path).ok()?;
|
||||
serde_json::from_str::<VaultFile>(&text)
|
||||
.ok()
|
||||
.map(|f| f.accounts)
|
||||
}
|
||||
|
||||
/// Move a damaged vault aside so it is preserved but out of the way, and
|
||||
/// report where it went.
|
||||
fn quarantine(path: &Path) -> Option<PathBuf> {
|
||||
let stamp = OffsetDateTime::now_utc().unix_timestamp();
|
||||
let dest = path.with_file_name(format!("vault.json.corrupt-{stamp}"));
|
||||
std::fs::rename(path, &dest).ok().map(|()| dest)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
@@ -678,6 +955,8 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn corrupt_vault_is_fatal_rather_than_silently_empty() {
|
||||
// No backup to fall back on, so this is the unrecoverable case and the
|
||||
// original refusal must stand.
|
||||
let dir = tempdir().unwrap();
|
||||
std::fs::write(dir.path().join("vault.json"), b"{ not json").unwrap();
|
||||
let r = Vault::load(dir.path(), VaultKey::from_bytes(&[1u8; 32]).unwrap());
|
||||
@@ -857,12 +1136,15 @@ mod tests {
|
||||
#[test]
|
||||
fn message_ids_are_persisted_as_strings() {
|
||||
// Snowflakes exceed 2^53; a JSON number would round-trip lossily.
|
||||
// Message ids now live in push-state.json, so that is where the
|
||||
// quoting has to hold.
|
||||
let dir = tempdir().unwrap();
|
||||
let v = vault(dir.path());
|
||||
let a = v.add(&sample(), None, None, true).unwrap();
|
||||
v.mark_pushed(&a.id, "1409876543210987654", "000000");
|
||||
v.flush_state_now();
|
||||
|
||||
let raw = std::fs::read_to_string(dir.path().join("vault.json")).unwrap();
|
||||
let raw = std::fs::read_to_string(dir.path().join("push-state.json")).unwrap();
|
||||
assert!(
|
||||
raw.contains("\"1409876543210987654\""),
|
||||
"must be quoted: {raw}"
|
||||
@@ -875,6 +1157,23 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_secret_bearing_file_stays_free_of_discord_state() {
|
||||
// The split is the protection: whatever goes wrong with the file that
|
||||
// is rewritten constantly, it cannot take the secrets with it.
|
||||
let dir = tempdir().unwrap();
|
||||
let v = vault(dir.path());
|
||||
let a = v.add(&sample(), None, None, true).unwrap();
|
||||
v.mark_pushed(&a.id, "1409876543210987654", "000000");
|
||||
v.flush_state_now();
|
||||
|
||||
let vault_json = std::fs::read_to_string(dir.path().join("vault.json")).unwrap();
|
||||
assert!(
|
||||
!vault_json.contains("1409876543210987654"),
|
||||
"vault.json must not carry tile ids: {vault_json}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn push_targets_exclude_paused_accounts_and_keep_order() {
|
||||
let dir = tempdir().unwrap();
|
||||
@@ -907,4 +1206,249 @@ mod tests {
|
||||
assert!(v.push_targets()[0].message_id.is_none());
|
||||
assert!(!v.get(&a.id).unwrap().push_live);
|
||||
}
|
||||
|
||||
// ── Recovery from a damaged vault ───────────────────────────────────────
|
||||
//
|
||||
// The failure these cover: an unclean shutdown left `vault.json` present
|
||||
// but zero bytes, and the old load path refused to start, so the container
|
||||
// restart-looped with no way back in.
|
||||
|
||||
#[test]
|
||||
fn a_zero_length_vault_recovers_from_the_backup() {
|
||||
let dir = tempdir().unwrap();
|
||||
let id = {
|
||||
let v = vault(dir.path());
|
||||
let id = v.add(&sample(), None, None, true).unwrap().id;
|
||||
// A second change is what creates the .bak of the first.
|
||||
v.add(&sample(), Some("Second"), None, true).unwrap();
|
||||
id
|
||||
};
|
||||
// Simulate the interrupted write: an empty file where the vault was.
|
||||
std::fs::write(dir.path().join("vault.json"), b"").unwrap();
|
||||
|
||||
let recovered = vault(dir.path());
|
||||
assert_eq!(recovered.len(), 2, "the backup must be used, not ignored");
|
||||
assert!(
|
||||
recovered.totp_for(&id).is_ok(),
|
||||
"recovered accounts must still decrypt"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_recovered_vault_keeps_the_damaged_file_for_inspection() {
|
||||
let dir = tempdir().unwrap();
|
||||
{
|
||||
let v = vault(dir.path());
|
||||
v.add(&sample(), None, None, true).unwrap();
|
||||
v.add(&sample(), Some("Second"), None, true).unwrap();
|
||||
}
|
||||
std::fs::write(dir.path().join("vault.json"), b"").unwrap();
|
||||
let _ = vault(dir.path());
|
||||
|
||||
let quarantined: Vec<_> = std::fs::read_dir(dir.path())
|
||||
.unwrap()
|
||||
.flatten()
|
||||
.map(|e| e.file_name().to_string_lossy().into_owned())
|
||||
.filter(|n| n.starts_with("vault.json.corrupt-"))
|
||||
.collect();
|
||||
assert_eq!(
|
||||
quarantined.len(),
|
||||
1,
|
||||
"the damaged file must be moved aside, not deleted: {quarantined:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_recovery_is_written_back_so_the_next_boot_is_clean() {
|
||||
let dir = tempdir().unwrap();
|
||||
{
|
||||
let v = vault(dir.path());
|
||||
v.add(&sample(), None, None, true).unwrap();
|
||||
v.add(&sample(), Some("Second"), None, true).unwrap();
|
||||
}
|
||||
std::fs::write(dir.path().join("vault.json"), b"").unwrap();
|
||||
|
||||
// First boot recovers; a later edit rewrites vault.json from memory.
|
||||
{
|
||||
let v = vault(dir.path());
|
||||
assert_eq!(v.len(), 2);
|
||||
v.add(&sample(), Some("Third"), None, true).unwrap();
|
||||
}
|
||||
let reloaded = vault(dir.path());
|
||||
assert_eq!(reloaded.len(), 3);
|
||||
assert_eq!(reloaded.push_targets().len(), 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_recovery_puts_the_vault_back_on_disk_immediately() {
|
||||
let dir = tempdir().unwrap();
|
||||
{
|
||||
let v = vault(dir.path());
|
||||
v.add(&sample(), None, None, true).unwrap();
|
||||
v.add(&sample(), Some("Second"), None, true).unwrap();
|
||||
}
|
||||
std::fs::write(dir.path().join("vault.json"), b"").unwrap();
|
||||
|
||||
// Booting must leave a real vault behind, not just the .bak. If it
|
||||
// did not, the recovered accounts would live only in memory and a
|
||||
// second unclean stop would have nothing left to recover from.
|
||||
{
|
||||
let v = vault(dir.path());
|
||||
assert_eq!(v.len(), 2);
|
||||
}
|
||||
let on_disk = std::fs::read_to_string(dir.path().join("vault.json")).unwrap();
|
||||
assert!(
|
||||
on_disk.contains("Cloudflare"),
|
||||
"the vault must be rebuilt on load: {on_disk}"
|
||||
);
|
||||
// And a second boot is then entirely ordinary.
|
||||
assert_eq!(vault(dir.path()).len(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_backup_that_is_also_unreadable_still_refuses_to_start() {
|
||||
let dir = tempdir().unwrap();
|
||||
std::fs::write(dir.path().join("vault.json"), b"not json").unwrap();
|
||||
std::fs::write(dir.path().join("vault.json.bak"), b"{ not json either").unwrap();
|
||||
|
||||
let r = Vault::load(dir.path(), VaultKey::from_bytes(&[1u8; 32]).unwrap());
|
||||
assert!(
|
||||
r.is_err(),
|
||||
"a backup that cannot be parsed is no recovery at all"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_fresh_install_is_not_an_error() {
|
||||
let dir = tempdir().unwrap();
|
||||
let v = vault(dir.path());
|
||||
assert!(v.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn valid_json_with_no_accounts_still_loads() {
|
||||
// An operator who deliberately emptied their vault must not be told
|
||||
// their file is corrupt.
|
||||
let dir = tempdir().unwrap();
|
||||
std::fs::write(dir.path().join("vault.json"), b"{\"accounts\":[]}").unwrap();
|
||||
let v = vault(dir.path());
|
||||
assert!(v.is_empty());
|
||||
}
|
||||
|
||||
// ── Discord bookkeeping lives outside the vault ─────────────────────────
|
||||
|
||||
#[test]
|
||||
fn rotating_a_code_does_not_rewrite_the_vault() {
|
||||
let dir = tempdir().unwrap();
|
||||
let v = vault(dir.path());
|
||||
let a = v.add(&sample(), None, None, true).unwrap();
|
||||
let vault_path = dir.path().join("vault.json");
|
||||
let before = std::fs::read(&vault_path).unwrap();
|
||||
|
||||
// A code rotation is a mark_pushed per account, thousands of times a
|
||||
// day. The file holding the secrets must not be one of the things
|
||||
// being rewritten.
|
||||
for code in ["111111", "222222", "333333"] {
|
||||
v.mark_pushed(&a.id, "1409876543210987654", code);
|
||||
}
|
||||
v.flush_state_now();
|
||||
|
||||
assert_eq!(
|
||||
std::fs::read(&vault_path).unwrap(),
|
||||
before,
|
||||
"vault.json must be untouched by push bookkeeping"
|
||||
);
|
||||
assert!(dir.path().join("push-state.json").exists());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn push_state_survives_a_reload() {
|
||||
let dir = tempdir().unwrap();
|
||||
let id = {
|
||||
let v = vault(dir.path());
|
||||
let a = v.add(&sample(), None, None, true).unwrap();
|
||||
v.mark_pushed(&a.id, "1409876543210987654", "000000");
|
||||
v.flush_state_now();
|
||||
a.id
|
||||
};
|
||||
|
||||
// Without this the restart would repost every tile in the channel,
|
||||
// because the engine would no longer know the messages exist.
|
||||
let reloaded = vault(dir.path());
|
||||
let target = &reloaded.push_targets()[0];
|
||||
assert_eq!(target.id, id);
|
||||
assert_eq!(
|
||||
target.message_id.as_deref(),
|
||||
Some("1409876543210987654"),
|
||||
"the tile id must be re-attached to the account"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_broken_push_state_file_costs_nothing_but_a_repost() {
|
||||
let dir = tempdir().unwrap();
|
||||
{
|
||||
let v = vault(dir.path());
|
||||
let a = v.add(&sample(), None, None, true).unwrap();
|
||||
v.mark_pushed(&a.id, "555", "000000");
|
||||
v.flush_state_now();
|
||||
}
|
||||
std::fs::write(dir.path().join("push-state.json"), b"not json at all").unwrap();
|
||||
|
||||
// Unlike vault.json, this must never stop the service starting.
|
||||
let reloaded = vault(dir.path());
|
||||
assert_eq!(reloaded.len(), 1);
|
||||
assert!(reloaded.push_targets()[0].message_id.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pending_state_is_not_written_on_every_push() {
|
||||
let dir = tempdir().unwrap();
|
||||
let v = vault(dir.path());
|
||||
let a = v.add(&sample(), None, None, true).unwrap();
|
||||
v.flush_state_now();
|
||||
let state_path = dir.path().join("push-state.json");
|
||||
|
||||
// A handful of rotations in quick succession must coalesce into one
|
||||
// write rather than one write each: that is the entire reason the
|
||||
// bookkeeping is batched. `flush_state` is what the push loop calls
|
||||
// once per tick, and it declines while the last write is still fresh.
|
||||
for code in ["111111", "222222", "333333"] {
|
||||
v.mark_pushed(&a.id, "999", code);
|
||||
v.flush_state();
|
||||
}
|
||||
let after_ticks = std::fs::read(&state_path).unwrap();
|
||||
assert_eq!(
|
||||
after_ticks,
|
||||
b"{\"accounts\":{}}",
|
||||
"nothing may reach disk yet, but got: {}",
|
||||
String::from_utf8_lossy(&after_ticks)
|
||||
);
|
||||
|
||||
// An explicit flush is the shutdown path, and must not defer.
|
||||
v.flush_state_now();
|
||||
let flushed = std::fs::read(&state_path).unwrap();
|
||||
assert!(
|
||||
flushed.windows(3).any(|w| w == b"999"),
|
||||
"the tile id must reach disk on an explicit flush: {}",
|
||||
String::from_utf8_lossy(&flushed)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn push_state_for_removed_accounts_is_dropped() {
|
||||
let dir = tempdir().unwrap();
|
||||
let v = vault(dir.path());
|
||||
let a = v.add(&sample(), None, None, true).unwrap();
|
||||
v.mark_pushed(&a.id, "555", "000000");
|
||||
v.flush_state_now();
|
||||
v.remove(&a.id).unwrap();
|
||||
v.flush_state_now();
|
||||
|
||||
let raw = std::fs::read_to_string(dir.path().join("push-state.json")).unwrap();
|
||||
assert!(
|
||||
!raw.contains(&a.id),
|
||||
"a state entry for a deleted account is dead weight: {raw}"
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -61,6 +61,10 @@ async fn main() -> anyhow::Result<()> {
|
||||
let admin = AdminStore::load_or_default(&args.work_dir);
|
||||
let settings = SettingsStore::load_or_default(&args.work_dir);
|
||||
|
||||
// Held separately from `state` so the shutdown path below can still reach
|
||||
// the vault after the router has taken ownership.
|
||||
let vault_on_shutdown = vault.clone();
|
||||
|
||||
let engine = PushEngine::new(vault.clone(), settings.clone());
|
||||
tokio::spawn(engine.clone().run());
|
||||
|
||||
@@ -97,6 +101,11 @@ async fn main() -> anyhow::Result<()> {
|
||||
.await
|
||||
.context("the HTTP server stopped unexpectedly")?;
|
||||
|
||||
// The push loop only flushes bookkeeping when a few seconds have piled up.
|
||||
// On a clean stop there is no next tick, so write whatever is still
|
||||
// pending rather than leaving the last rotation to be rediscovered.
|
||||
vault_on_shutdown.flush_state_now();
|
||||
|
||||
tracing::info!(target: "da", "shutting down");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -58,6 +58,12 @@ impl std::fmt::Debug for PushEngine {
|
||||
}
|
||||
|
||||
impl PushEngine {
|
||||
/// The vault this engine is driving, so a caller shutting down can flush
|
||||
/// the bookkeeping it has been batching.
|
||||
pub fn vault(&self) -> &Vault {
|
||||
&self.vault
|
||||
}
|
||||
|
||||
pub fn new(vault: Vault, settings: Arc<SettingsStore>) -> Self {
|
||||
Self {
|
||||
vault,
|
||||
@@ -81,6 +87,14 @@ impl PushEngine {
|
||||
/// One pass: bring every tile up to date and tidy away any that shouldn't
|
||||
/// exist. Returns how many messages were written.
|
||||
pub async fn sync_once(&self) -> usize {
|
||||
let written = self.sync_tiles().await;
|
||||
// Bookkeeping changes are batched in memory; this is the tick that
|
||||
// decides whether enough of them have piled up to be worth a write.
|
||||
self.vault.flush_state();
|
||||
written
|
||||
}
|
||||
|
||||
async fn sync_tiles(&self) -> usize {
|
||||
// Held for the whole pass: see `sync_lock`. A pass can take several
|
||||
// seconds once the rate limiter spaces things out, which is easily
|
||||
// long enough for the next tick to arrive.
|
||||
|
||||
Reference in new issue
Block a user