randogoth/CalCalist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add aggregate retarget

Browse source 
Moving an aggregate to a different endpoint is the sharpest failure mode in the
design, so it is a command rather than something a config edit triggers: sync
already refuses on target drift and points here.

The move settles against the old target first. An event a user created in the
aggregate and that has not yet reached a sink exists only there, and would be
lost the moment we stop looking at that calendar. Only then is every mapped event
rebuilt on the new target — a rewrite rather than a copy, since the rendered
content differs between backends that can and cannot suppress scheduling.

Orphans on the old target are kept unless --purge-old, and a purge is bounded by
the derivation, so events the user keeps in that calendar themselves survive.

Also fixed prepare_vdirs, which created a directory per endpoint. That told
pimsync a local collection existed before its remote counterpart had been seen,
so it tried to create the counterpart: unsupported for a read-only feed, and it
would have invented calendars on a CalDAV server. Only the vdir root is created
now; collections are pimsync's to make from what it discovers.

One test premise was wrong rather than the code: a foreign event in an aggregate
that has a sink is a user-created event and gets adopted into that sink, so it
never reaches a purge as an outsider. Both behaviours are now covered.

Verified against the live Posteo calendar after the change: still a clean no-op.

106 tests.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
randogoth 2026-09-10 13:04:41 +03:00
parent 9f14c1f651
commit dd984e150a
4 changed files with 495 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

6
TODO.md
View file

@ -41,7 +41,7 @@ Safety-critical behaviour:
update and delete under `sendUpdates=none`. The Google attendee path depends on
it. Fallback if it fails: the same demotion transform used for CalDAV.
- [x] `sync` refuses to run on aggregate target drift, before reconciliation
- [ ] `aggregate retarget` — flush unrouted creations against the old target, then
- [x] `aggregate retarget` — flush unrouted creations against the old target, then
re-materialise; keep old orphans by default
- [x] Mass-deletion guard (`max_delete_fraction`), overridable with `--force`, with an
absolute floor so deleting a couple of events is never refused
@ -57,8 +57,8 @@ Tests:
`VALARM` intact, `PARTSTAT: DECLINED` maps to `TRANSP: TRANSPARENT`, bulk deletion aborts
- [ ] Safety (integration): the same against a real Radicale instance with an SMTP sink,
proving no mail is emitted
- [ ] Retarget: drift makes `sync` exit non-zero having written nothing and losing no source
event (verified by hand end to end; still needs an automated test)
- [x] Retarget: drift makes `sync` exit non-zero having written nothing and losing no source
event; purge is bounded by the derivation; an unrouted creation reaches a sink first
## M2 — interface and packaging

45
src/main.rs
View file

@ -10,6 +10,7 @@ mod paths;
mod pimsync;
mod provenance;
mod reconcile;
mod retarget;
mod state;
mod sync;
mod vdir;
@ -18,7 +19,7 @@ use std::process::ExitCode;
use clap::Parser;
use crate::cli::{Cli, Command, GoogleCommand};
use crate::cli::{AggregateCommand, Cli, Command, GoogleCommand};
use crate::config::Config;
use crate::google::auth;
use crate::sync::Report;
@ -40,6 +41,16 @@ fn main() -> ExitCode {
},
..
} => run_google_login(cli, endpoint),
cli @ Cli {
command:
Command::Aggregate {
command:
AggregateCommand::Retarget {
id, to, purge_old, ..
},
},
..
} => run_retarget(cli, id, to, *purge_old),
Cli { command, .. } => unimplemented(command),
}
}
@ -94,6 +105,38 @@ fn run_google_login(cli: &Cli, endpoint_id: &str) -> ExitCode {
}
}
/// Moves an aggregate to a different target endpoint, deliberately.
fn run_retarget(cli: &Cli, id: &str, to: &str, purge_old: bool) -> ExitCode {
let (config, _) = match Config::load(cli.config.as_deref()) {
Ok(loaded) => loaded,
Err(error) => return fail(&error),
};
let state_dir = match paths::state_dir() {
Ok(dir) => dir,
Err(error) => return fail(&error),
};
match retarget::retarget(&config, &state_dir, id, to, purge_old) {
Ok(outcome) => {
println!(
"{}: moved from `{}` to `{}` — {} event(s) rebuilt, {} flushed to a sink first, {} removed from the old target",
outcome.aggregate,
outcome.from,
outcome.to,
outcome.materialised,
outcome.routed_first,
outcome.purged,
);
if outcome.purged == 0 && !purge_old {
println!(
" the previous target still holds this aggregate's events; re-run with --purge-old to remove them"
);
}
ExitCode::SUCCESS
}
Err(error) => fail(&error),
}
}
fn print_report(report: &Report) {
if report.aggregates.is_empty() {
println!("no aggregates configured");

436
src/retarget.rs Normal file
View file

@ -0,0 +1,436 @@
//! Moving an aggregate to a different target endpoint.
//!
//! This is the sharpest failure mode in the design, which is why it is a command
//! of its own rather than something a config edit triggers. A plain `sync` after
//! the target changed would read the new, empty calendar as an aggregate whose
//! every event had been deleted — and with delete propagation on, remove them
//! from every source. `sync` therefore refuses on target drift and sends the user
//! here.
//!
//! Two properties of the design make the move mechanical rather than a rebuild:
//! the aggregate is a derived replica, so the sources stay authoritative
//! throughout, and UID derivation excludes the target, so every derived UID is
//! unchanged by the move. Only the rendered content and its location change.
use std::path::Path;
use thiserror::Error;
use crate::config::{Config, Endpoint};
use crate::mirror;
use crate::provenance;
use crate::state::{State, Target};
use crate::sync::{self, SyncError, vdir_path};
use crate::vdir::{self, VdirError};
#[derive(Debug, Error)]
pub enum RetargetError {
#[error(transparent)]
Sync(#[from] SyncError),
#[error(transparent)]
Vdir(#[from] VdirError),
#[error(transparent)]
State(#[from] crate::state::StateError),
#[error("no aggregate named `{0}` in the configuration")]
UnknownAggregate(String),
#[error("no endpoint named `{0}` in the configuration")]
UnknownEndpoint(String),
#[error("endpoint `{0}` is read-only, so an aggregate cannot be published to it")]
NotWritable(String),
#[error(
"endpoint `{endpoint}` is a source of aggregate `{aggregate}`, so targeting it would sync the aggregate into itself"
)]
TargetIsSource { endpoint: String, aggregate: String },
#[error(
"aggregate `{0}` has never been synced, so there is nothing to move; edit the config and run `calcalist sync`"
)]
NeverSynced(String),
}
#[derive(Debug)]
pub struct Outcome {
pub aggregate: String,
pub from: String,
pub to: String,
/// Events flushed to a sink before the move, which existed nowhere else.
pub routed_first: usize,
pub materialised: usize,
pub purged: usize,
}
/// Moves `aggregate_id` to `new_target_id`.
pub fn retarget(
config: &Config,
state_dir: &Path,
aggregate_id: &str,
new_target_id: &str,
purge_old: bool,
) -> Result<Outcome, RetargetError> {
let aggregate = config
.aggregates
.iter()
.find(|candidate| candidate.id == aggregate_id)
.ok_or_else(|| RetargetError::UnknownAggregate(aggregate_id.to_string()))?;
let new_target = endpoint(config, new_target_id)?;
if !new_target.kind.is_writable() {
return Err(RetargetError::NotWritable(new_target_id.to_string()));
}
if aggregate
.sources
.iter()
.any(|source| source == new_target_id)
{
return Err(RetargetError::TargetIsSource {
endpoint: new_target_id.to_string(),
aggregate: aggregate_id.to_string(),
});
}
let state_path = state_dir.join(crate::state::FILE_NAME);
let mut state = State::load(&state_path)?;
let recorded = state
.aggregate(aggregate_id)
.map(|entry| entry.target.clone())
.ok_or_else(|| RetargetError::NeverSynced(aggregate_id.to_string()))?;
let old_target = endpoint(config, &recorded.endpoint)?.clone();
sync::prepare_vdirs(config, state_dir)?;
let pimsync_config = if sync::needs_pimsync(config) {
Some(crate::pimsync::write_config(config, state_dir).map_err(SyncError::from)?)
} else {
None
};
if let Some(path) = &pimsync_config {
crate::pimsync::sync(path).map_err(SyncError::from)?;
}
// Settle against the old target first. Events a user created in the
// aggregate and that have not yet reached a sink exist only there, and would
// be lost the moment we stop looking at it.
let settled = sync::sync_aggregate(
config,
aggregate,
&old_target,
state_dir,
&mut state,
false,
false,
)?;
let materialised = materialise(config, aggregate, new_target, state_dir, &mut state)?;
let purged = if purge_old {
purge(aggregate_id, &old_target, state_dir, &state)?
} else {
0
};
let entry = state.aggregate_mut(aggregate_id, target_of(new_target));
entry.target = target_of(new_target);
state.save(&state_path)?;
if let Some(path) = &pimsync_config {
crate::pimsync::sync(path).map_err(SyncError::from)?;
}
Ok(Outcome {
aggregate: aggregate_id.to_string(),
from: old_target.id.clone(),
to: new_target.id.clone(),
routed_first: settled.written_back,
materialised,
purged,
})
}
/// Rebuilds every mapped event on the new target, using that backend's transform.
///
/// The rendered content differs between backends — a target that can suppress
/// scheduling keeps real attendees, one that cannot has them demoted — so this is
/// a rewrite of every event rather than a copy.
fn materialise(
config: &Config,
aggregate: &crate::config::Aggregate,
new_target: &Endpoint,
state_dir: &Path,
state: &mut State,
) -> Result<usize, RetargetError> {
let new_dir = vdir_path(state_dir, &new_target.id);
let existing = vdir::read(&new_dir)?;
let suppression = new_target.kind.scheduling_suppression();
let links = state
.aggregate(&aggregate.id)
.map(|entry| entry.links.clone())
.unwrap_or_default();
let mut written = 0;
let mut updated = links.clone();
for (aggregate_uid, link) in &links {
let source = config.endpoint(&link.source_id);
let Some(source) = source else { continue };
let source_items = vdir::read(&vdir_path(state_dir, &source.id))?;
let Some(item) = source_items.get(&link.source_uid) else {
// The source event is gone; the next ordinary sync will retire it.
continue;
};
let rebuilt = mirror::to_aggregate(
&item.calendar,
aggregate_uid,
&source.id,
&link.source_uid,
source.kind.owner(),
suppression,
);
let replaces = existing.get(aggregate_uid).map(|held| held.path.clone());
vdir::write(&new_dir, aggregate_uid, &rebuilt, replaces.as_deref())?;
if let Some(entry) = updated.get_mut(aggregate_uid) {
entry.aggregate_hash = rebuilt.content_hash();
}
written += 1;
}
state
.aggregate_mut(&aggregate.id, target_of(new_target))
.links = updated;
Ok(written)
}
/// Removes this aggregate's mirrors from the calendar it has left.
///
/// Bounded by the derivation: only events whose UID this aggregate would have
/// produced are touched, so anything the user keeps in that calendar survives.
fn purge(
aggregate_id: &str,
old_target: &Endpoint,
state_dir: &Path,
state: &State,
) -> Result<usize, RetargetError> {
let old_dir = vdir_path(state_dir, &old_target.id);
let items = vdir::read(&old_dir)?;
let links = state
.aggregate(aggregate_id)
.map(|entry| &entry.links)
.ok_or_else(|| RetargetError::NeverSynced(aggregate_id.to_string()))?;
let mut removed = 0;
for (uid, item) in &items {
if provenance::is_derived(uid) && links.contains_key(uid) {
vdir::remove(&item.path)?;
removed += 1;
}
}
Ok(removed)
}
/// The state record describing where an aggregate is published.
fn target_of(endpoint: &Endpoint) -> Target {
Target {
endpoint: endpoint.id.clone(),
kind: endpoint.kind.kind_name().to_string(),
}
}
fn endpoint<'a>(config: &'a Config, id: &str) -> Result<&'a Endpoint, RetargetError> {
config
.endpoint(id)
.ok_or_else(|| RetargetError::UnknownEndpoint(id.to_string()))
}
#[cfg(test)]
mod tests {
use super::*;
use std::fs;
/// Google endpoints are used as stand-ins: they need no pimsync, so a whole
/// cycle runs against local directories with nothing to reach over a network.
const CONFIG: &str = r#"
version = 1
[[endpoint]]
id = "src"
type = "google"
calendar_id = "src@example.com"
client_id = "x.apps.googleusercontent.com"
[[endpoint]]
id = "agg1"
type = "google"
calendar_id = "agg1@example.com"
client_id = "x.apps.googleusercontent.com"
[[endpoint]]
id = "agg2"
type = "google"
calendar_id = "agg2@example.com"
client_id = "x.apps.googleusercontent.com"
[[aggregate]]
id = "unified"
target = "agg1"
sources = ["src"]
default_sink = "src"
"#;
fn config() -> Config {
toml::from_str(CONFIG).expect("test config should parse")
}
fn event(uid: &str, summary: &str) -> String {
format!(
"BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\nUID:{uid}\r\nDTSTART:20260910T090000Z\r\nDTSTAMP:20260101T000000Z\r\nSUMMARY:{summary}\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n"
)
}
/// A state directory with two source events already mirrored into `agg1`.
fn synced() -> tempfile::TempDir {
let dir = tempfile::tempdir().expect("temp dir");
let src = vdir_path(dir.path(), "src");
fs::create_dir_all(&src).expect("create src vdir");
fs::write(src.join("a.ics"), event("a@example.com", "A")).expect("write");
fs::write(src.join("b.ics"), event("b@example.com", "B")).expect("write");
sync::run(&config(), dir.path(), false, false).expect("initial sync");
dir
}
fn count(dir: &Path, endpoint: &str) -> usize {
vdir::read(&vdir_path(dir, endpoint))
.expect("read vdir")
.len()
}
#[test]
fn rebuilds_every_event_on_the_new_target() {
let dir = synced();
assert_eq!(count(dir.path(), "agg1"), 2);
let outcome = retarget(&config(), dir.path(), "unified", "agg2", false).expect("retarget");
assert_eq!(outcome.materialised, 2);
assert_eq!(count(dir.path(), "agg2"), 2);
// Kept by default: the old calendar may be wanted as a snapshot.
assert_eq!(count(dir.path(), "agg1"), 2);
assert_eq!(outcome.purged, 0);
}
#[test]
fn records_the_new_target_so_sync_stops_refusing() {
let dir = synced();
retarget(&config(), dir.path(), "unified", "agg2", false).expect("retarget");
let state = State::load(&dir.path().join(crate::state::FILE_NAME)).expect("state");
assert_eq!(
state.aggregate("unified").expect("entry").target.endpoint,
"agg2"
);
}
/// A purge is bounded by the derivation, so a calendar the user also keeps
/// their own events in does not lose them.
///
/// The aggregate has no sink here deliberately: with one configured, a
/// foreign event in the aggregate is a user-created event and gets adopted
/// into the sink rather than staying foreign, so it would never reach the
/// purge as an outsider.
#[test]
fn purging_removes_only_this_aggregates_mirrors() {
let sinkless: Config = toml::from_str(&CONFIG.replace(r#"default_sink = "src""#, ""))
.expect("config should parse");
let dir = tempfile::tempdir().expect("temp dir");
let src = vdir_path(dir.path(), "src");
fs::create_dir_all(&src).expect("create src vdir");
fs::write(src.join("a.ics"), event("a@example.com", "A")).expect("write");
fs::write(src.join("b.ics"), event("b@example.com", "B")).expect("write");
sync::run(&sinkless, dir.path(), false, false).expect("initial sync");
let old = vdir_path(dir.path(), "agg1");
fs::write(old.join("mine.ics"), event("my-own@example.com", "Mine")).expect("write");
assert_eq!(count(dir.path(), "agg1"), 3);
let outcome = retarget(&sinkless, dir.path(), "unified", "agg2", true).expect("retarget");
assert_eq!(outcome.purged, 2, "only the aggregate's own mirrors go");
let remaining = vdir::read(&old).expect("read");
assert_eq!(remaining.len(), 1);
assert!(
remaining.contains_key("my-own@example.com"),
"the user's own event was destroyed"
);
}
/// The converse, and the reason the test above needs a sinkless aggregate:
/// an event created in an aggregate that has a sink is adopted into it.
#[test]
fn a_foreign_event_is_adopted_when_a_sink_exists() {
let dir = synced();
let old = vdir_path(dir.path(), "agg1");
fs::write(old.join("mine.ics"), event("my-own@example.com", "Mine")).expect("write");
retarget(&config(), dir.path(), "unified", "agg2", true).expect("retarget");
let sources = vdir::read(&vdir_path(dir.path(), "src")).expect("read");
assert!(sources.contains_key("my-own@example.com"));
}
/// An event created in the aggregate and not yet routed exists nowhere else,
/// so the move must flush it to a sink before it stops looking at that
/// calendar.
#[test]
fn an_unrouted_creation_reaches_the_sink_before_the_move() {
let dir = synced();
let old = vdir_path(dir.path(), "agg1");
fs::write(old.join("hand.ics"), event("hand@phone", "Handwritten")).expect("write");
let outcome = retarget(&config(), dir.path(), "unified", "agg2", true).expect("retarget");
assert_eq!(outcome.routed_first, 1);
let sources = vdir::read(&vdir_path(dir.path(), "src")).expect("read");
assert!(
sources.contains_key("hand@phone"),
"the only copy of the event was lost"
);
}
#[test]
fn refuses_a_target_that_is_one_of_the_sources() {
let dir = tempfile::tempdir().expect("temp dir");
let error =
retarget(&config(), dir.path(), "unified", "src", false).expect_err("should refuse");
assert!(
matches!(error, RetargetError::TargetIsSource { .. }),
"{error}"
);
}
/// Writability is checked before any state is touched, so this needs no
/// prior sync — which keeps a webcal endpoint out of the shared fixture.
#[test]
fn refuses_a_read_only_target() {
let with_feed: Config = toml::from_str(&format!(
"{CONFIG}\n[[endpoint]]\nid = \"feed\"\ntype = \"webcal\"\nurl = \"https://example.org/f.ics\"\n"
))
.expect("config should parse");
let dir = tempfile::tempdir().expect("temp dir");
let error =
retarget(&with_feed, dir.path(), "unified", "feed", false).expect_err("should refuse");
assert!(matches!(error, RetargetError::NotWritable(_)), "{error}");
}
#[test]
fn refuses_unknown_names() {
let dir = tempfile::tempdir().expect("temp dir");
assert!(matches!(
retarget(&config(), dir.path(), "nope", "agg2", false).expect_err("refuse"),
RetargetError::UnknownAggregate(_)
));
assert!(matches!(
retarget(&config(), dir.path(), "unified", "nowhere", false).expect_err("refuse"),
RetargetError::UnknownEndpoint(_)
));
}
/// Nothing has been published yet, so there is nothing to move.
#[test]
fn refuses_an_aggregate_that_has_never_synced() {
let dir = tempfile::tempdir().expect("temp dir");
let error =
retarget(&config(), dir.path(), "unified", "agg2", false).expect_err("should refuse");
assert!(matches!(error, RetargetError::NeverSynced(_)), "{error}");
}
}

22
src/sync.rs
View file

@ -118,7 +118,7 @@ pub fn run(
}
/// Whether any endpoint is one pimsync handles. A Google-only setup needs none.
fn needs_pimsync(config: &Config) -> bool {
pub(crate) fn needs_pimsync(config: &Config) -> bool {
config.endpoints.iter().any(|endpoint| {
matches!(
endpoint.kind,
@ -127,14 +127,16 @@ fn needs_pimsync(config: &Config) -> bool {
})
}
/// pimsync requires every vdir collection to exist before it will run, and an
/// endpoint that has never synced has no directory yet.
fn prepare_vdirs(config: &Config, state_dir: &Path) -> Result<(), SyncError> {
for endpoint in &config.endpoints {
let dir = vdir_path(state_dir, &endpoint.id);
std::fs::create_dir_all(&dir).map_err(|source| SyncError::Prepare { path: dir, source })?;
}
Ok(())
/// pimsync requires its vdir storage directory to exist before it will run.
///
/// Only the root is created. Creating a directory per endpoint would tell
/// pimsync that a local collection exists before its remote counterpart has been
/// seen, and it would then try to create that counterpart — which a read-only
/// feed cannot do, and which would silently invent calendars on a CalDAV server.
/// Collections are pimsync's to create from what it discovers.
pub(crate) fn prepare_vdirs(_config: &Config, state_dir: &Path) -> Result<(), SyncError> {
let root = vdir_root(state_dir);
std::fs::create_dir_all(&root).map_err(|source| SyncError::Prepare { path: root, source })
}
fn resolve<'a>(
@ -176,7 +178,7 @@ fn check_target_drift(
})
}
fn sync_aggregate(
pub(crate) fn sync_aggregate(
config: &Config,
aggregate: &Aggregate,
target: &Endpoint,