From dd984e150a5fd4a3adfa0bfafcd8f3cfc42e62f0 Mon Sep 17 00:00:00 2001 From: randogoth Date: Thu, 10 Sep 2026 13:04:41 +0300 Subject: [PATCH] Add aggregate retarget MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- TODO.md | 6 +- src/main.rs | 45 ++++- src/retarget.rs | 436 ++++++++++++++++++++++++++++++++++++++++++++++++ src/sync.rs | 22 +-- 4 files changed, 495 insertions(+), 14 deletions(-) create mode 100644 src/retarget.rs diff --git a/TODO.md b/TODO.md index 720d555..65f152e 100644 --- a/TODO.md +++ b/TODO.md @@ -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 diff --git a/src/main.rs b/src/main.rs index 8a300f6..108abfd 100644 --- a/src/main.rs +++ b/src/main.rs @@ -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"); diff --git a/src/retarget.rs b/src/retarget.rs new file mode 100644 index 0000000..99f8771 --- /dev/null +++ b/src/retarget.rs @@ -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 { + 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 { + 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 { + 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}"); + } +} diff --git a/src/sync.rs b/src/sync.rs index c996b35..b1f20e3 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -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,