src/sync_cycle.rs
Rama main · réplica de lectura del repositorio original.
//! Single-cycle sync runner.
//!
//! This module owns the per-cycle boundary that sits between the outer
//! process/watch loop and the download engine: mode choice, enumeration config
//! hash handling, grouping preload, download dispatch, zone-token writes, and
//! the aggregate cycle result.
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use crate::auth;
use crate::config;
use crate::download;
use crate::state;
/// Per-library state: zone name, sync token key, and resolved album plan.
pub(crate) struct LibraryState {
pub(crate) library: crate::icloud::photos::PhotoLibrary,
pub(crate) cross_zone_libraries: Vec<crate::icloud::photos::PhotoLibrary>,
pub(crate) pass_scope: crate::commands::PassScope,
pub(crate) zone_name: String,
pub(crate) sync_token_key: String,
/// Ordered list of download passes. Each pass carries its own
/// exclude-asset-ids set. See [`crate::commands::AlbumPlan`].
pub(crate) plan: crate::commands::AlbumPlan,
/// True when `resolve_passes` failed at the end of the prior cycle and
/// the plan above is the previous cycle's stale snapshot. Album
/// membership data captured under a stale plan can route assets to the
/// wrong pass (e.g. an asset added to a newly-created album shows up in
/// the unfiled pass), so any cycle that consumes a stale plan must not
/// advance the sync token for any zone -- doing so would skip the
/// change events those assets generated and leave `asset_albums`
/// permanently incomplete.
pub(crate) plan_is_stale: bool,
/// True after an idle watch sleep. Refreshing only when a later
/// `changes/database` pre-check finds relevant work avoids burning album
/// listing calls on quiet watch cycles.
pub(crate) plan_needs_refresh: bool,
}
/// Metadata key holding the SHA-256 of the enumeration-affecting subset of
/// the user's download config. Distinct from the path-affecting
/// `config_hash` consumed by the download pipeline; using a single key for
/// both would cause each cycle to overwrite the other's value and
/// permanently invalidate incremental sync.
pub(crate) const ENUM_CONFIG_HASH_KEY: &str = "enum_config_hash";
pub(crate) const PENDING_ENUM_CONFIG_HASH_KEY: &str = "pending_enum_config_hash";
pub(crate) const PENDING_DOWNLOAD_CONFIG_HASH_KEY: &str = "pending_download_config_hash";
const PENDING_ZONE_TOKEN_PREFIX: &str = "pending_sync_token:";
const LAST_CHECKPOINT_STATUS_KEY: &str = "last_checkpoint_status";
const LAST_RECOVERY_ACTION_KEY: &str = "last_recovery_action";
const LAST_FULL_ENUMERATION_REASON_KEY: &str = "last_full_enumeration_reason";
/// Prefix for every per-zone CloudKit sync token row in the metadata table.
pub(crate) const SYNC_TOKEN_PREFIX: &str = "sync_token:";
// Legacy inventory_anchor: rows lack coverage proof and are never read or migrated.
const INVENTORY_ANCHOR_PREFIX: &str = "inventory_anchor_v1:";
const INVENTORY_ANCHOR_VERSION: u8 = 1;
/// Presence is a durable obligation, including unknown future payloads. Only a
/// proven FINISHED full inventory may remove it with checkpoint promotion.
pub(crate) const INDEXING_RECONCILIATION_PREFIX: &str = "indexing_reconciliation_required:";
pub(crate) fn indexing_reconciliation_key(zone: &str) -> String {
format!("{INDEXING_RECONCILIATION_PREFIX}{zone}")
}
/// A qualified inventory retains the old cursor and schedules its replay in a
/// later trusted cycle. No deletion-capable bridge runs inside the held phase.
pub(crate) const INDEXING_REPLAY_PREFIX: &str = "indexing_replay_required:";
pub(crate) fn indexing_replay_key(zone: &str) -> String {
format!("{INDEXING_REPLAY_PREFIX}{zone}")
}
/// Managed qualification did not queue files. This obligation forces a full
/// managed publication pass before retained-cursor replay may discharge it.
pub(crate) fn indexing_materialization_key(zone: &str) -> String {
format!("indexing_materialization_required:{zone}")
}
/// Observe readiness and durably retain every RUNNING observation, even if a
/// later readiness request fails before enumeration starts.
pub(crate) async fn prepare_indexing_reconciliation(
library: &crate::icloud::photos::PhotoLibrary,
db: Option<&dyn download::DownloadStore>,
dry_run: bool,
) -> anyhow::Result<(bool, Option<String>)> {
let key = indexing_reconciliation_key(library.zone_name());
let mut hold = match db {
Some(db) => db.get_metadata(&key).await?,
None => None,
};
if !library.last_indexing_finished() && !dry_run {
let db = db.ok_or_else(|| {
anyhow::anyhow!("Provisional indexing downloads require durable state")
})?;
if hold.is_none() {
db.set_metadata(&key, "1").await?;
hold = Some("1".to_owned());
}
}
let finished = library.check_indexing().await?;
if !finished && !dry_run {
let db = db.ok_or_else(|| {
anyhow::anyhow!("Provisional indexing downloads require durable state")
})?;
if hold.is_none() {
db.set_metadata(&key, "1").await?;
hold = Some("1".to_owned());
}
}
Ok((finished, hold))
}
/// Cross-zone album hydration contributes to the selected listing too. Any
/// unfinished source makes the selected zone provisional until all are ready.
pub(crate) async fn prepare_library_indexing(
library: &LibraryState,
db: Option<&dyn download::DownloadStore>,
dry_run: bool,
) -> anyhow::Result<(bool, Option<String>)> {
let (mut finished, mut hold) =
prepare_indexing_reconciliation(&library.library, db, dry_run).await?;
for source in &library.cross_zone_libraries {
if source.zone_name() == library.zone_name {
continue;
}
let source_was_running = !source.last_indexing_finished();
let (source_finished, _) = prepare_indexing_reconciliation(source, db, dry_run).await?;
finished &= source_finished;
if (source_was_running || !source_finished) && hold.is_none() && !dry_run {
let db = db.ok_or_else(|| {
anyhow::anyhow!("Provisional indexing downloads require durable state")
})?;
db.set_metadata(&indexing_reconciliation_key(&library.zone_name), "1")
.await?;
hold = Some("1".to_owned());
}
}
Ok((finished, hold))
}
pub(crate) const RETAINED_CHECKPOINT_HOLD_PREFIX: &str = "retained_checkpoint_hold:";
const RETAINED_CHECKPOINT_RETRY_SECONDS: i64 = 60 * 60;
const RETAINED_CHECKPOINT_MAX_ATTEMPTS: u8 = 3;
const fn first_retained_checkpoint_attempt() -> u8 {
1
}
/// A delay is bound to the retained boundary and the requested policy. It never
/// certifies inventory coverage or historical identity/deletion resolution.
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct RetainedCheckpointHold {
version: u8,
prior_cursor_sha256: String,
enum_config_hash: String,
preservation_config_hash: Option<String>,
next_attempt_at: i64,
#[serde(default = "first_retained_checkpoint_attempt")]
attempt_count: u8,
}
impl RetainedCheckpointHold {
fn new(prior_cursor: &str, enum_hash: &str, preservation_hash: Option<&str>) -> Self {
use sha2::{Digest, Sha256};
Self {
version: 1,
attempt_count: 1,
prior_cursor_sha256: data_encoding::HEXLOWER.encode(&Sha256::digest(prior_cursor)),
enum_config_hash: enum_hash.to_owned(),
preservation_config_hash: preservation_hash.map(str::to_owned),
next_attempt_at: chrono::Utc::now()
.timestamp()
.saturating_add(RETAINED_CHECKPOINT_RETRY_SECONDS),
}
}
fn matches(&self, current: &Self) -> bool {
self.version == 1
&& self.attempt_count > 0
&& self.prior_cursor_sha256 == current.prior_cursor_sha256
&& self.enum_config_hash == current.enum_config_hash
&& self.preservation_config_hash == current.preservation_config_hash
}
fn deferred(&self, now: i64) -> bool {
self.next_attempt_at > now
&& self.next_attempt_at <= now.saturating_add(RETAINED_CHECKPOINT_RETRY_SECONDS)
}
}
pub(crate) fn retained_checkpoint_hold_action(encoded: &str) -> &'static str {
match serde_json::from_str::<RetainedCheckpointHold>(encoded) {
Ok(hold) if hold.version == 1 && hold.attempt_count > 0 => {
if hold.attempt_count >= RETAINED_CHECKPOINT_MAX_ATTEMPTS {
RecoveryAction::RetainedCheckpointRetryExhausted.as_str()
} else {
RecoveryAction::AwaitRetainedCheckpointEvidence.as_str()
}
}
_ => RecoveryAction::RepairRetainedCheckpointEvidence.as_str(),
}
}
fn retained_checkpoint_hold_key(zone: &str) -> String {
format!("{RETAINED_CHECKPOINT_HOLD_PREFIX}{zone}")
}
fn block_for_retained_checkpoint(stats: &mut download::SyncStats) {
stats.sync_token_blocked = true;
stats.sync_token_blocked_reason = Some("retained_checkpoint_expired");
stats.sync_token_blocked_source = Some("kei");
stats.sync_token_blocked_explanation = Some(download::sync_token_blocked_explanation(
"retained_checkpoint_expired",
));
}
async fn persist_retained_checkpoint_hold(
db: &dyn download::DownloadStore,
key: String,
hold: &RetainedCheckpointHold,
expected_metadata: Vec<(String, Option<String>)>,
) -> anyhow::Result<()> {
db.commit_checkpoint_transition(state::CheckpointTransition {
expected_metadata,
legacy_preservation_proofs: Vec::new(),
legacy_config_hash: None,
sparse_identity_proofs: Vec::new(),
metadata_updates: vec![
(key, serde_json::to_string(hold)?),
(
LAST_CHECKPOINT_STATUS_KEY.to_owned(),
"preserved".to_owned(),
),
(
LAST_RECOVERY_ACTION_KEY.to_owned(),
RecoveryAction::AwaitRetainedCheckpointEvidence
.as_str()
.to_owned(),
),
],
metadata_deletes: Vec::new(),
})
.await?;
Ok(())
}
const INVENTORY_DROP_THRESHOLD_PERCENT: f64 = 5.0;
#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
struct InventoryAnchor {
version: u8,
scope: download::InventoryScope,
api_total_at_start: u64,
assets_seen: u64,
completed_at: i64,
}
#[derive(Debug, Clone, Copy, PartialEq)]
struct InventoryDrop {
previous_total: u64,
current_total: u64,
drop_assets: u64,
drop_percent: f64,
}
/// Return the metadata key that stores the CloudKit sync token for a zone.
pub(crate) fn sync_token_key(zone_name: &str) -> String {
format!("{SYNC_TOKEN_PREFIX}{zone_name}")
}
fn inventory_anchor_key(
enum_config_hash: &str,
zone_name: &str,
scope: &download::InventoryScope,
) -> String {
use sha2::{Digest, Sha256};
let scope_hash = data_encoding::HEXLOWER.encode(&Sha256::digest(scope.0.as_bytes()));
format!("{INVENTORY_ANCHOR_PREFIX}{enum_config_hash}:{zone_name}:{scope_hash}")
}
pub(crate) fn pending_zone_token_key(enum_config_hash: &str, zone_name: &str) -> String {
format!("{PENDING_ZONE_TOKEN_PREFIX}{enum_config_hash}:{zone_name}")
}
#[allow(
clippy::cast_precision_loss,
reason = "inventory totals are operator diagnostics; f64 percent is only for warning/report context"
)]
fn classify_inventory_drop(previous_total: u64, current_total: u64) -> Option<InventoryDrop> {
if previous_total == 0 || current_total >= previous_total {
return None;
}
let drop_assets = previous_total - current_total;
let drop_percent = (drop_assets as f64 / previous_total as f64) * 100.0;
if drop_percent < INVENTORY_DROP_THRESHOLD_PERCENT {
return None;
}
Some(InventoryDrop {
previous_total,
current_total,
drop_assets,
drop_percent,
})
}
/// Outcome of a single sync cycle across all libraries.
#[derive(Debug)]
pub(crate) struct CycleResult {
pub(crate) failed_count: usize,
pub(crate) session_expired: bool,
pub(crate) stats: download::SyncStats,
pub(crate) db_sync_token_advance_safe: bool,
}
impl CycleResult {
/// The database pre-check needs a clean aggregate cycle as well as safe
/// selected-zone checkpoints. A durable transfer failure can satisfy the
/// zone gate without satisfying this broader gate.
#[must_use]
pub(crate) fn can_advance_database_checkpoint(&self) -> bool {
!self.session_expired
&& self.failed_count == 0
&& !self.stats.interrupted
&& self.db_sync_token_advance_safe
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CheckpointBasis {
IncrementalDelta,
CompleteInventory,
InventoryWithDeltaBridge,
}
impl CheckpointBasis {
const fn as_str(self) -> &'static str {
match self {
Self::IncrementalDelta => "incremental_delta",
Self::CompleteInventory => "complete_inventory",
Self::InventoryWithDeltaBridge => "inventory_with_delta_bridge",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CheckpointHoldReason {
DryRun,
StalePassPlan,
Interrupted,
SessionExpired,
EnumerationIncomplete,
StateNotDurable,
TokenProofIncomplete,
LegacyPreservationIncomplete,
}
impl CheckpointHoldReason {
const fn as_str(self) -> &'static str {
match self {
Self::DryRun => "dry_run",
Self::StalePassPlan => "stale_pass_plan",
Self::Interrupted => "interrupted",
Self::SessionExpired => "session_expired",
Self::EnumerationIncomplete => "enumeration_incomplete",
Self::StateNotDurable => "state_not_durable",
Self::TokenProofIncomplete => "token_proof_incomplete",
Self::LegacyPreservationIncomplete => "legacy_preservation_incomplete",
}
}
}
pub(crate) use download::RecoveryAction;
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum SourceCheckpointDecision {
Advance {
token: String,
basis: CheckpointBasis,
},
Preserve {
reason: CheckpointHoldReason,
recovery: RecoveryAction,
},
}
/// Capture the owning decision without serializing its provider token.
pub(crate) fn checkpoint_support_fields(decision: &SourceCheckpointDecision) -> serde_json::Value {
match decision {
SourceCheckpointDecision::Advance { basis, .. } => serde_json::json!({
"decision":"advanced", "basis":basis.as_str(), "recovery":"none"
}),
SourceCheckpointDecision::Preserve { reason, recovery } => serde_json::json!({
"decision":"preserved", "reason":reason.as_str(), "recovery":recovery.as_str()
}),
}
}
fn observe_sync_component(
zone: &str,
component: &'static str,
result: &download::SyncResult,
passes: usize,
) {
crate::support::observe_scoped(
zone,
None,
"support_pass_completion_v1",
serde_json::json!({
"component":component, "count_scope":"selected_invocation", "passes":passes,
"phase":if result.full_enumeration_ran {"full"} else {"incremental"},
// These totals describe this invocation, before any bridge composition.
// Query EOF evidence is recorded by the provider fetcher, not inferred
// from absence of reported errors or widened into inventory coverage.
"inventory_complete":serde_json::Value::Null,
"selection_comparable":serde_json::Value::Null,
"enumeration_incomplete":result.checkpoint.enumeration_incomplete,
"enumeration_errors":result.checkpoint.enumeration_errors,
"identity_incomplete":result.checkpoint.identity_incomplete,
"interrupted":result.checkpoint.interrupted,
"api_total":result.stats.api_total_at_start, "assets_seen":result.stats.assets_seen,
"outcome":match result.outcome {
download::DownloadOutcome::Success => "success",
download::DownloadOutcome::SessionExpired { .. } => "session_expired",
download::DownloadOutcome::PartialFailure { .. } => "partial_failure",
},
"token_present":result.sync_token.is_some(),
"token_usable":result.sync_token.as_ref().is_some_and(|token| !token.trim().is_empty()),
"sync_token_blocked":result.checkpoint.sync_token_blocked,
"sync_token_blocked_reason":result.stats.sync_token_blocked_reason,
"sync_token_blocked_source":result.stats.sync_token_blocked_source,
"sync_token_expected_receivers":result.stats.sync_token_expected_receivers,
"sync_token_receivers_with_token":result.stats.sync_token_receivers_with_token,
"sync_token_receivers_missing":result.stats.sync_token_receivers_missing,
"sync_token_receivers_blank":result.stats.sync_token_receivers_blank,
"sync_token_receivers_dropped":result.stats.sync_token_receivers_dropped,
"sync_token_unique_values":result.stats.sync_token_unique_values,
"same_cycle_recovery_attempts":result.stats.same_cycle_recovery_attempts,
"same_cycle_recovery_successes":result.stats.same_cycle_recovery_successes,
"completed_delta_replay":result.checkpoint.completed_delta_replay,
"retained_token_rejected":result.checkpoint.retained_token_rejected,
}),
);
}
fn source_checkpoint_decision(
result: &download::SyncResult,
dry_run: bool,
stale_pass_plan: bool,
basis: CheckpointBasis,
) -> SourceCheckpointDecision {
// CONTRACT: SOURCE_CHECKPOINT_REQUIRES_DURABLE_RECOVERY
let preserve = |reason, recovery| SourceCheckpointDecision::Preserve { reason, recovery };
if dry_run {
return preserve(CheckpointHoldReason::DryRun, RecoveryAction::Stop);
}
if stale_pass_plan {
return preserve(
CheckpointHoldReason::StalePassPlan,
RecoveryAction::ReplayFromPriorToken,
);
}
if matches!(
result.outcome,
download::DownloadOutcome::SessionExpired { .. }
) {
return preserve(
CheckpointHoldReason::SessionExpired,
RecoveryAction::Reauthenticate,
);
}
if result.checkpoint.interrupted {
return preserve(
CheckpointHoldReason::Interrupted,
RecoveryAction::ReplayFromPriorToken,
);
}
if result.checkpoint.enumeration_errors > 0 || result.checkpoint.enumeration_incomplete {
return preserve(
CheckpointHoldReason::EnumerationIncomplete,
RecoveryAction::ContinueTail,
);
}
if result.checkpoint.state_write_failures > 0 {
return preserve(
CheckpointHoldReason::StateNotDurable,
RecoveryAction::ReplayFromPriorToken,
);
}
let Some(token) = result
.sync_token
.as_ref()
.filter(|token| !token.trim().is_empty())
else {
let recovery = if !result.checkpoint.retry_passes.is_empty() {
RecoveryAction::RetryPasses(result.checkpoint.retry_passes.clone())
} else if !result.checkpoint.revalidate_records.is_empty() {
RecoveryAction::RevalidateRecords(result.checkpoint.revalidate_records.clone())
} else {
RecoveryAction::ReplayFromPriorToken
};
return preserve(CheckpointHoldReason::TokenProofIncomplete, recovery);
};
if result.checkpoint.sync_token_blocked || result.checkpoint.identity_incomplete {
let recovery = if result.checkpoint.retry_passes.is_empty() {
RecoveryAction::ReplayFromPriorToken
} else {
RecoveryAction::RetryPasses(result.checkpoint.retry_passes.clone())
};
return preserve(CheckpointHoldReason::TokenProofIncomplete, recovery);
}
SourceCheckpointDecision::Advance {
token: token.clone(),
basis,
}
}
fn merge_download_outcomes(
first: &download::DownloadOutcome,
second: &download::DownloadOutcome,
) -> download::DownloadOutcome {
match (first, second) {
(download::DownloadOutcome::SessionExpired { auth_error_count }, _)
| (_, download::DownloadOutcome::SessionExpired { auth_error_count }) => {
download::DownloadOutcome::SessionExpired {
auth_error_count: *auth_error_count,
}
}
(
download::DownloadOutcome::PartialFailure {
failed_count: first,
},
download::DownloadOutcome::PartialFailure {
failed_count: second,
},
) => download::DownloadOutcome::PartialFailure {
failed_count: first.saturating_add(*second),
},
(download::DownloadOutcome::PartialFailure { failed_count }, _)
| (_, download::DownloadOutcome::PartialFailure { failed_count }) => {
download::DownloadOutcome::PartialFailure {
failed_count: *failed_count,
}
}
(download::DownloadOutcome::Success, download::DownloadOutcome::Success) => {
download::DownloadOutcome::Success
}
}
}
/// Decide whether the per-zone `sync_token` should be persisted to the state
/// DB after a download pass.
///
/// The contract is "advance only on full success and not in dry-run":
/// - On `PartialFailure`, a stored token would skip the failed assets on the
/// next incremental sync (they'd never appear in the delta again -- silent
/// data loss).
/// - On `SessionExpired`, the cycle aborts mid-stream; the token may be
/// stale or only reflect a subset of the work.
/// - In `--dry-run`, we promise to make no DB writes that survive the run
/// (apart from the `sync_runs` ledger). Advancing the token here would
/// silently break the next real sync.
///
/// The returned bool is the gate: callers still check that `sync_token` is
/// `Some(_)` and that a state DB is configured before persisting.
#[cfg(test)]
pub(crate) fn should_store_sync_token(outcome: &download::DownloadOutcome, dry_run: bool) -> bool {
matches!(outcome, download::DownloadOutcome::Success) && !dry_run
}
/// Library-level gate that combines the per-library outcome check with the
/// stale-plan flag for the same zone.
///
/// A reused album plan can route assets created or moved between cycles to
/// the wrong pass, so the affected zone must not advance. Unaffected zones
/// can still store their own zone tokens; the broader database pre-check
/// token remains conservative until every selected zone is clean.
#[cfg(test)]
pub(crate) fn should_store_sync_token_for_cycle(
outcome: &download::DownloadOutcome,
dry_run: bool,
library_plan_is_stale: bool,
) -> bool {
should_store_sync_token(outcome, dry_run) && !library_plan_is_stale
}
fn has_active_passes(lib_state: &LibraryState) -> bool {
!lib_state.plan.passes.is_empty()
}
fn should_warn_zero_assets(
sync_result: &download::SyncResult,
library_completed_without_errors: bool,
run_mode: download::DownloadRunMode,
is_retry_failed: bool,
lib_state: &LibraryState,
) -> bool {
sync_result.full_enumeration_ran
&& library_completed_without_errors
&& run_mode.downloads_files()
&& !is_retry_failed
&& sync_result.stats.assets_seen == 0
&& has_active_passes(lib_state)
}
async fn update_inventory_anchor_for_cycle<D>(
db: &D,
enum_config_hash: &str,
zone_name: &str,
inventory: &download::InventoryMeasurement,
stats: &mut download::SyncStats,
) where
D: state::SyncTokenStore + ?Sized,
{
let current_total = inventory.total;
let key = inventory_anchor_key(enum_config_hash, zone_name, &inventory.scope);
match db.get_metadata(&key).await {
Ok(Some(raw)) => match serde_json::from_str::<InventoryAnchor>(&raw) {
Ok(anchor)
if anchor.version == INVENTORY_ANCHOR_VERSION
&& anchor.scope == inventory.scope =>
{
if let Some(drop) =
classify_inventory_drop(anchor.api_total_at_start, current_total)
{
tracing::warn!(
library = zone_name,
previous_api_total = drop.previous_total,
current_api_total = drop.current_total,
drop_assets = drop.drop_assets,
drop_percent = drop.drop_percent,
threshold_percent = INVENTORY_DROP_THRESHOLD_PERCENT,
strict_inventory = false,
enum_config_hash,
"Inventory dropped below previous comparable full run"
);
stats.inventory_drop_warnings = stats.inventory_drop_warnings.saturating_add(1);
stats.inventory_drop_assets = drop.drop_assets;
stats.inventory_drop_percent = Some(drop.drop_percent);
stats.inventory_drop_previous_total = Some(drop.previous_total);
stats.inventory_drop_current_total = Some(drop.current_total);
stats.inventory_drop_library = Some(zone_name.to_string());
}
}
Ok(_) => {}
Err(e) => {
tracing::debug!(
error = %e,
library = zone_name,
"Ignoring unreadable inventory anchor"
);
}
},
Ok(None) => {}
Err(e) => {
tracing::debug!(
error = %e,
library = zone_name,
"Failed to read inventory anchor"
);
return;
}
}
let anchor = InventoryAnchor {
version: INVENTORY_ANCHOR_VERSION,
scope: inventory.scope.clone(),
api_total_at_start: current_total,
assets_seen: inventory.assets_seen,
completed_at: chrono::Utc::now().timestamp(),
};
match serde_json::to_string(&anchor) {
Ok(value) => {
if let Err(e) = db.set_metadata(&key, &value).await {
tracing::debug!(
error = %e,
library = zone_name,
"Failed to persist inventory anchor"
);
}
}
Err(e) => {
tracing::debug!(
error = %e,
library = zone_name,
"Failed to serialize inventory anchor"
);
}
}
}
/// Closure shape used to derive a per-library `DownloadConfig` from the
/// shared base config. Boxed dyn so `run_cycle` can accept a single
/// reference instead of a generic parameter (avoids reuse-by-monomorphization
/// blow-up in error messages).
pub(crate) type BuildDownloadConfigFn<'a> = dyn Fn(
download::SyncMode,
Arc<rustc_hash::FxHashSet<String>>,
Arc<download::AssetGroupings>,
Arc<str>,
) -> Arc<download::DownloadConfig>
+ 'a;
/// Outcome of [`check_and_persist_enum_config_hash`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum EnumConfigHashOutcome {
/// No prior hash; current hash persisted. Sync tokens left alone:
/// a first-run DB must not invalidate tokens another process may
/// have written.
Initial,
Unchanged,
/// Hash drifted; the candidate hash was staged while active tokens and
/// the active hash remain unchanged until reconciliation succeeds.
Changed,
/// Hash drift was detected, but the pending reconciliation hash could
/// not be staged.
ChangedTokenPurgeFailed,
/// The stored hash could not be read, so the cycle cannot prove whether
/// existing sync tokens still match the current config.
ReadFailed,
}
impl EnumConfigHashOutcome {
fn must_force_full_sync(self) -> bool {
matches!(
self,
Self::Changed | Self::ChangedTokenPurgeFailed | Self::ReadFailed
)
}
}
/// State of the path-affecting download config hash before a cycle.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum DownloadConfigHashOutcome {
Unchanged,
Changed,
ReadFailed,
TokenPurgeFailed,
}
impl DownloadConfigHashOutcome {
const fn must_force_full_sync(self) -> bool {
matches!(self, Self::ReadFailed | Self::TokenPurgeFailed)
}
const fn token_purge_failed(self) -> bool {
matches!(self, Self::TokenPurgeFailed)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SyncModeDecision {
pub(crate) mode: download::SyncMode,
pub(crate) full_enumeration_reason: Option<download::FullEnumerationReason>,
}
fn metadata_refresh_mode_decision(refresh_metadata: bool) -> Option<SyncModeDecision> {
refresh_metadata.then_some(SyncModeDecision {
mode: download::SyncMode::Full,
full_enumeration_reason: Some(download::FullEnumerationReason::MetadataBackfill),
})
}
/// Compare the current enumeration-config hash against the active value.
/// Drift is staged without deleting the last safe provider checkpoint.
pub(crate) async fn check_and_persist_enum_config_hash<D>(
db: &D,
current_hash: &str,
) -> EnumConfigHashOutcome
where
D: state::SyncTokenStore + ?Sized,
{
let stored_hash = match db.get_metadata(ENUM_CONFIG_HASH_KEY).await {
Ok(hash) => hash,
Err(e) => {
tracing::warn!(error = %e, "Failed to read enum_config_hash");
return EnumConfigHashOutcome::ReadFailed;
}
};
let outcome = match stored_hash.as_deref() {
Some(h) if h == current_hash => {
if let Err(e) = db
.delete_metadata_by_prefix(PENDING_ENUM_CONFIG_HASH_KEY)
.await
{
tracing::debug!(error = %e, "Failed to clear reverted pending enum config hash");
}
if let Err(e) = db
.delete_metadata_by_prefix(PENDING_ZONE_TOKEN_PREFIX)
.await
{
tracing::debug!(error = %e, "Failed to clear reverted pending zone checkpoints");
}
return EnumConfigHashOutcome::Unchanged;
}
Some(_) => EnumConfigHashOutcome::Changed,
None => EnumConfigHashOutcome::Initial,
};
let key = if matches!(outcome, EnumConfigHashOutcome::Changed) {
tracing::info!(
"Enumeration config changed since last sync; preserving active checkpoints and staging reconciliation"
);
match db.get_metadata(PENDING_ENUM_CONFIG_HASH_KEY).await {
Ok(Some(pending_hash)) if pending_hash == current_hash => {}
Ok(_) => {
if let Err(e) = db
.delete_metadata_by_prefix(PENDING_ZONE_TOKEN_PREFIX)
.await
{
tracing::warn!(error = %e, "Failed to clear checkpoints from superseded config reconciliation");
return EnumConfigHashOutcome::ChangedTokenPurgeFailed;
}
}
Err(e) => {
tracing::warn!(error = %e, "Failed to read pending enum config hash");
return EnumConfigHashOutcome::ReadFailed;
}
}
PENDING_ENUM_CONFIG_HASH_KEY
} else {
ENUM_CONFIG_HASH_KEY
};
if let Err(e) = db.set_metadata(key, current_hash).await {
tracing::warn!(error = %e, key, "Failed to persist enum config hash state");
if matches!(outcome, EnumConfigHashOutcome::Changed) {
return EnumConfigHashOutcome::ChangedTokenPurgeFailed;
}
}
outcome
}
/// Check the path-affecting download config hash without deleting provider
/// cursors. Reconciliation promotes the pending hash only after it succeeds.
///
/// This hash does not prove CloudKit cursor safety. It proves local
/// path/state trust: after a directory or filename-template change, an
/// incremental `/changes/zone` cycle would only see new deltas and could miss
/// already-known assets that must be written to the new local paths.
pub(crate) async fn check_download_config_hash_for_cycle<D>(
db: &D,
current_hash: &str,
legacy_hash: &str,
) -> DownloadConfigHashOutcome
where
D: state::SyncTokenStore + ?Sized,
{
match db.get_metadata(download::DOWNLOAD_CONFIG_HASH_KEY).await {
Ok(Some(stored)) if stored == current_hash => {
if let Err(e) = db
.delete_metadata_by_prefix(PENDING_DOWNLOAD_CONFIG_HASH_KEY)
.await
{
tracing::debug!(error = %e, "Failed to clear reverted pending download config hash");
}
DownloadConfigHashOutcome::Unchanged
}
Ok(Some(stored)) if stored == legacy_hash => {
if let Err(e) = db
.commit_checkpoint_transition(state::CheckpointTransition {
expected_metadata: Vec::new(),
legacy_preservation_proofs: Vec::new(),
legacy_config_hash: None,
sparse_identity_proofs: Vec::new(),
metadata_updates: vec![(
download::DOWNLOAD_CONFIG_HASH_KEY.to_owned(),
current_hash.to_owned(),
)],
metadata_deletes: vec![PENDING_DOWNLOAD_CONFIG_HASH_KEY.to_owned()],
})
.await
{
tracing::warn!(error = %e, "Failed to migrate legacy download config_hash");
return DownloadConfigHashOutcome::ReadFailed;
}
tracing::info!("Migrated legacy download path config hash");
DownloadConfigHashOutcome::Unchanged
}
Ok(None) => {
if let Err(e) = db
.set_metadata(download::DOWNLOAD_CONFIG_HASH_KEY, current_hash)
.await
{
tracing::warn!(error = %e, "Failed to persist download config_hash");
return DownloadConfigHashOutcome::ReadFailed;
}
DownloadConfigHashOutcome::Unchanged
}
Ok(Some(_stored)) => {
tracing::info!(
"Download path config changed since last sync; preserving provider checkpoints and staging local reconciliation"
);
if let Err(e) = db
.set_metadata(PENDING_DOWNLOAD_CONFIG_HASH_KEY, current_hash)
.await
{
tracing::warn!(error = %e, "Failed to stage pending download config_hash");
DownloadConfigHashOutcome::TokenPurgeFailed
} else {
if let Err(e) = db
.set_metadata(LAST_RECOVERY_ACTION_KEY, "reconcile_local_paths")
.await
{
tracing::debug!(error = %e, "Failed to persist local path reconciliation state");
}
DownloadConfigHashOutcome::Changed
}
}
Err(e) => {
tracing::warn!(error = %e, "Failed to read download config_hash");
DownloadConfigHashOutcome::ReadFailed
}
}
}
/// Run one sync cycle: iterate all libraries, download photos, store sync tokens.
pub(crate) async fn run_cycle(
library_states: &[&LibraryState],
config: &config::Config,
state_db: Option<&dyn download::DownloadStore>,
is_retry_failed: bool,
build_download_config: &BuildDownloadConfigFn<'_>,
download_controls: download::DownloadControls,
shared_session: &auth::SharedSession,
shutdown_token: &CancellationToken,
) -> anyhow::Result<CycleResult> {
let mut cycle_failed_count = 0usize;
let mut cycle_session_expired = false;
let mut cycle_stats = download::SyncStats::default();
let mut sparse_identity_proofs = Vec::new();
let cycle_has_stale_plan = library_states.iter().any(|s| s.plan_is_stale);
if cycle_has_stale_plan {
tracing::warn!(
"One or more libraries are running on a stale album plan; sync \
database pre-check token will not advance this cycle"
);
}
// CONTRACT: SYNC_TOKEN_ADVANCE_REQUIRES_CLEAN_CYCLE
let mut db_sync_token_advance_safe = !config.runtime.dry_run && !cycle_has_stale_plan;
let mut force_full_for_config_hash = false;
let mut force_full_for_download_config_hash = false;
let mut checkpoint_transition_state_safe = true;
let mut enum_config_hash_outcome = EnumConfigHashOutcome::Unchanged;
let mut pending_download_config_hash = None;
let mut path_reconciliation_complete = true;
let mut path_reconciliation_cycle_safe = true;
let path_reconciliation_has_smart_folders = library_states.iter().any(|state| {
state
.plan
.passes
.iter()
.any(|pass| pass.kind == crate::commands::PassKind::SmartFolder)
});
let mut checkpoint_hold_action: Option<&'static str> = None;
let enum_config_hash = download::compute_config_hash(config);
// Check if token-unsafe eligibility config changed since last sync. If
// so, clear sync tokens and force full enumeration for this cycle -- the
// stored incremental token would miss assets that are newly eligible
// under the changed config (e.g. a user switching [photos].resolution or
// changing [filters].media). The hash is
// cycle-invariant across libraries,
// so this runs once per cycle, not once per library.
//
// The metadata key `enum_config_hash` is distinct from the download
// pipeline's `config_hash` (which tracks path-affecting fields only).
// Using a single key for both would cause the two hashes to overwrite
// each other every cycle, permanently preventing incremental sync.
if !config.runtime.dry_run
&& let Some(db) = state_db
{
enum_config_hash_outcome = check_and_persist_enum_config_hash(db, &enum_config_hash).await;
force_full_for_config_hash = enum_config_hash_outcome.must_force_full_sync();
if matches!(enum_config_hash_outcome, EnumConfigHashOutcome::Changed) {
let recovery = RecoveryAction::ReconcileInventory(
download::FullEnumerationReason::EnumConfigHashDrift,
);
if let Err(e) = db
.set_metadata(LAST_RECOVERY_ACTION_KEY, recovery.as_str())
.await
{
tracing::debug!(error = %e, "Failed to persist inventory reconciliation action");
}
}
if matches!(
enum_config_hash_outcome,
EnumConfigHashOutcome::ChangedTokenPurgeFailed | EnumConfigHashOutcome::ReadFailed
) {
checkpoint_transition_state_safe = false;
db_sync_token_advance_safe = false;
}
}
let mut preservation_config_hash = None;
// A path-affecting config drift does not make the CloudKit token itself
// unsafe, but it does make an incremental cycle insufficient: with no
// changed assets, `/changes/zone` would never re-plan already-known media
// into the new directory/template. Detect that before choosing
// incremental mode. Path-only drift keeps source tracking incremental;
// the pending marker drives separate catalog/targeted rehydration work.
if !config.runtime.dry_run
&& let (Some(db), Some(first_library)) = (state_db, library_states.first())
{
let probe_config = build_download_config(
download::SyncMode::Full,
Arc::new(rustc_hash::FxHashSet::default()),
Arc::new(download::AssetGroupings::default()),
Arc::from(first_library.zone_name.as_str()),
);
let download_config_hash = download::hash_download_config(&probe_config);
{
let mut zones = library_states
.iter()
.map(|library| library.zone_name.clone())
.collect::<Vec<_>>();
zones.sort();
zones.dedup();
let coverage = download::sync_coverage_fingerprint_json(
config,
"icloud",
1,
&zones,
&enum_config_hash,
&download_config_hash,
)?;
let policy = serde_json::json!([
1,
state::METADATA_CAPTURE_REVISION,
coverage,
format!("{:?}", probe_config.metadata),
config.runtime.refresh_metadata,
config.runtime.repair_capture_timestamps,
config.runtime.repair_truncated
]);
preservation_config_hash = Some(hash_legacy_preservation_policy(
&policy,
probe_config.legacy_preservation_allow_hardlinks,
));
}
let legacy_download_config_hash = download::hash_legacy_download_config(&probe_config);
let outcome = check_download_config_hash_for_cycle(
db,
&download_config_hash,
&legacy_download_config_hash,
)
.await;
force_full_for_download_config_hash = outcome.must_force_full_sync();
if matches!(outcome, DownloadConfigHashOutcome::Changed) {
pending_download_config_hash = Some(download_config_hash);
}
if outcome.token_purge_failed() {
checkpoint_transition_state_safe = false;
db_sync_token_advance_safe = false;
}
}
let path_metadata_at_start = if let Some(db) = state_db {
vec![
(
download::DOWNLOAD_CONFIG_HASH_KEY.to_owned(),
db.get_metadata(download::DOWNLOAD_CONFIG_HASH_KEY).await?,
),
(
PENDING_DOWNLOAD_CONFIG_HASH_KEY.to_owned(),
pending_download_config_hash.clone(),
),
]
} else {
Vec::new()
};
let enum_metadata_at_start = if let Some(db) = state_db {
vec![
(
ENUM_CONFIG_HASH_KEY.to_owned(),
db.get_metadata(ENUM_CONFIG_HASH_KEY).await?,
),
(
PENDING_ENUM_CONFIG_HASH_KEY.to_owned(),
db.get_metadata(PENDING_ENUM_CONFIG_HASH_KEY).await?,
),
]
} else {
Vec::new()
};
let mut active_cursor_fences = Vec::new();
for lib_state in library_states.iter().copied() {
if shutdown_token.is_cancelled() {
break;
}
let indexing_key = indexing_reconciliation_key(&lib_state.zone_name);
let (indexing_finished, indexing_hold_at_start) =
prepare_library_indexing(lib_state, state_db, config.runtime.dry_run).await?;
let indexing_reconciliation = indexing_hold_at_start.is_some() || !indexing_finished;
if !has_active_passes(lib_state) {
db_sync_token_advance_safe &= !indexing_reconciliation;
continue;
}
if !indexing_finished {
db_sync_token_advance_safe = false;
path_reconciliation_complete = false;
tracing::warn!(
"Photo library indexing is RUNNING; downloads are provisional, deletions and checkpoint promotion are disabled"
);
}
let retained_cursor_at_start = if let Some(db) = state_db {
db.get_metadata(&lib_state.sync_token_key).await?
} else {
None
};
let mut retained_checkpoint_attempt = 1;
let hold_key = retained_checkpoint_hold_key(&lib_state.zone_name);
let retained_hold_at_start = if let Some(db) = state_db {
db.get_metadata(&hold_key).await?
} else {
None
};
let replay_key = indexing_replay_key(&lib_state.zone_name);
let replay_at_start = match state_db {
Some(db) => db.get_metadata(&replay_key).await?,
None => None,
};
let indexing_replay_pending = replay_at_start.is_some();
let materialization_key = indexing_materialization_key(&lib_state.zone_name);
let materialization_at_start = match state_db {
Some(db) => db.get_metadata(&materialization_key).await?,
None => None,
};
let indexing_materialization_pending = materialization_at_start.is_some();
let mut zone_metadata_fence = enum_metadata_at_start.clone();
zone_metadata_fence.push((
materialization_key.clone(),
materialization_at_start.clone(),
));
active_cursor_fences.push((materialization_key.clone(), materialization_at_start));
zone_metadata_fence.push((indexing_key.clone(), indexing_hold_at_start.clone()));
zone_metadata_fence.push((replay_key.clone(), replay_at_start.clone()));
active_cursor_fences.push((replay_key.clone(), replay_at_start));
active_cursor_fences.push((indexing_key.clone(), indexing_hold_at_start.clone()));
zone_metadata_fence.push((
lib_state.sync_token_key.clone(),
retained_cursor_at_start.clone(),
));
active_cursor_fences.push((
lib_state.sync_token_key.clone(),
retained_cursor_at_start.clone(),
));
zone_metadata_fence.push((hold_key.clone(), retained_hold_at_start.clone()));
// Successful zone staging removes its hold before aggregate activation.
active_cursor_fences.push((hold_key.clone(), None));
if !config.runtime.dry_run
&& retained_hold_at_start.as_deref().is_some_and(|encoded| {
retained_checkpoint_hold_action(encoded)
== RecoveryAction::RepairRetainedCheckpointEvidence.as_str()
})
{
cycle_stats.sync_token_blocked = true;
cycle_stats.sync_token_blocked_reason = Some("retained_checkpoint_hold_invalid");
cycle_stats.sync_token_blocked_source = Some("kei");
cycle_stats.sync_token_blocked_explanation = Some(
download::sync_token_blocked_explanation("retained_checkpoint_hold_invalid"),
);
cycle_stats
.sync_token_blocked_zone
.get_or_insert_with(|| lib_state.zone_name.clone());
cycle_failed_count = cycle_failed_count.saturating_add(1);
db_sync_token_advance_safe = false;
checkpoint_hold_action =
Some(RecoveryAction::RepairRetainedCheckpointEvidence.as_str());
tracing::warn!(
diagnostic = "retained_checkpoint_hold_invalid",
"Retained checkpoint evidence requires validation; preserving the durable hold"
);
continue;
}
// A rejected retained boundary cannot be repaired by repeating the same
// rank inventory. Reserve each due attempt durably before provider or
// legacy work, including attempts later cancelled or found incomplete.
if !config.runtime.dry_run
&& let Some(db) = state_db
&& let Some(encoded) = retained_hold_at_start
&& let Ok(hold) = serde_json::from_str::<RetainedCheckpointHold>(&encoded)
&& let Some(prior) = retained_cursor_at_start.as_deref()
{
let mut current = RetainedCheckpointHold::new(
prior,
&enum_config_hash,
preservation_config_hash.as_deref(),
);
if hold.matches(¤t) {
let exhausted = hold.attempt_count >= RETAINED_CHECKPOINT_MAX_ATTEMPTS;
current.attempt_count = hold.attempt_count.saturating_add(1);
retained_checkpoint_attempt = current.attempt_count;
let deferred = exhausted || hold.deferred(chrono::Utc::now().timestamp());
let reservation_failed = if deferred {
false
} else {
persist_retained_checkpoint_hold(
db,
hold_key.clone(),
¤t,
zone_metadata_fence.clone(),
)
.await
.is_err()
};
if !deferred
&& !reservation_failed
&& let Some((_, expected)) = zone_metadata_fence.last_mut()
{
*expected = Some(serde_json::to_string(¤t)?);
}
if deferred || reservation_failed {
block_for_retained_checkpoint(&mut cycle_stats);
cycle_stats
.sync_token_blocked_zone
.get_or_insert_with(|| lib_state.zone_name.clone());
if reservation_failed {
cycle_stats.state_write_failures =
cycle_stats.state_write_failures.saturating_add(1);
}
cycle_failed_count = cycle_failed_count.saturating_add(1);
db_sync_token_advance_safe = false;
checkpoint_hold_action = Some(if exhausted {
RecoveryAction::RetainedCheckpointRetryExhausted.as_str()
} else {
RecoveryAction::AwaitRetainedCheckpointEvidence.as_str()
});
tracing::warn!(
diagnostic = "retained_checkpoint_expired",
retry_deferred = deferred,
state_write_failed = reservation_failed,
retry_exhausted = exhausted,
"Retained checkpoint evidence is unavailable; preserving history and deferring recovery"
);
continue;
}
}
}
let mut legacy_cycle = download::legacy_preservation::LegacyCycle::default();
if indexing_finished
&& !indexing_reconciliation
&& download_controls.run_mode.downloads_files()
&& !lib_state.plan_is_stale
&& let Some(db) = state_db
&& db
.get_summary()
.await?
.metadata_capture
.iter()
.any(|capture| {
capture.library == lib_state.zone_name
&& (capture.unresolved_assets > 0 || capture.unattributed_legacy_assets > 0)
})
{
let legacy_probe = build_download_config(
download::SyncMode::Full,
Arc::new(rustc_hash::FxHashSet::default()),
Arc::new(download::AssetGroupings::default()),
Arc::from(lib_state.zone_name.as_str()),
);
legacy_cycle = download::legacy_preservation::LegacyCycle::begin(
db,
&legacy_probe,
&lib_state.library.all(),
preservation_config_hash.as_deref().unwrap_or(""),
shutdown_token,
)
.await?;
}
// Determine source-enumeration mode per library. Failed transfers are
// rehydrated from durable pending state by the download engine and do
// not require replaying the provider inventory.
let pending_zone_token =
if matches!(enum_config_hash_outcome, EnumConfigHashOutcome::Changed) {
if let Some(db) = state_db {
db.get_metadata(&pending_zone_token_key(
&enum_config_hash,
&lib_state.zone_name,
))
.await?
.filter(|token| !token.trim().is_empty())
} else {
None
}
} else {
None
};
let sync_mode_decision = if indexing_reconciliation || indexing_materialization_pending {
SyncModeDecision {
mode: download::SyncMode::Full,
full_enumeration_reason: Some(download::FullEnumerationReason::OtherStaticReason),
}
} else if let Some(refresh_decision) =
metadata_refresh_mode_decision(config.runtime.refresh_metadata)
{
refresh_decision
} else if legacy_cycle.requires_inventory() {
SyncModeDecision {
mode: download::SyncMode::Full,
full_enumeration_reason: Some(download::FullEnumerationReason::MetadataBackfill),
}
} else if let Some(zone_sync_token) = pending_zone_token {
tracing::debug!(
zone = %lib_state.zone_name,
"Continuing config reconciliation from the completed zone checkpoint"
);
SyncModeDecision {
mode: download::SyncMode::Incremental { zone_sync_token },
full_enumeration_reason: None,
}
} else if force_full_for_config_hash {
let reason = download::FullEnumerationReason::EnumConfigHashDrift;
tracing::debug!(
zone = %lib_state.zone_name,
full_enumeration_reason = reason.as_str(),
"Forcing full sync because config-hash validation invalidated stored sync tokens"
);
SyncModeDecision {
mode: download::SyncMode::Full,
full_enumeration_reason: Some(reason),
}
} else if force_full_for_download_config_hash {
let reason = download::FullEnumerationReason::DownloadConfigHashDrift;
tracing::debug!(
zone = %lib_state.zone_name,
full_enumeration_reason = reason.as_str(),
"Forcing full sync because download config hash drift requires local path reconciliation"
);
SyncModeDecision {
mode: download::SyncMode::Full,
full_enumeration_reason: Some(reason),
}
} else {
determine_sync_mode_decision(
is_retry_failed,
library_states.len(),
state_db,
&lib_state.sync_token_key,
&lib_state.zone_name,
)
.await
};
let sync_mode = sync_mode_decision.mode;
let sync_mode_label = match &sync_mode {
download::SyncMode::Full => "full",
download::SyncMode::Incremental { .. } => "incremental",
};
if let Some(reason) = sync_mode_decision.full_enumeration_reason {
if let Some(db) = state_db
&& let Err(e) = db
.set_metadata(LAST_FULL_ENUMERATION_REASON_KEY, reason.as_str())
.await
{
tracing::debug!(error = %e, "Failed to persist full-enumeration reason");
}
tracing::info!(
sync_mode = sync_mode_label,
zone = %lib_state.zone_name,
full_enumeration_reason = reason.as_str(),
"Starting full-enumeration sync cycle"
);
} else {
tracing::debug!(sync_mode = sync_mode_label, zone = %lib_state.zone_name, "Starting sync cycle");
}
// Skip the DB scan entirely when nothing downstream will read it.
#[cfg(feature = "xmp")]
let asset_groupings = if config.metadata.embed_xmp || config.metadata.xmp_sidecar {
preload_asset_groupings(state_db, &lib_state.zone_name).await
} else {
Arc::new(download::AssetGroupings::default())
};
#[cfg(not(feature = "xmp"))]
let asset_groupings = Arc::new(download::AssetGroupings::default());
// Each pass carries its own exclude-asset-ids, so the config built
// here starts with an empty set; download_photos_with_sync derives
// per-pass configs internally via `with_exclude_ids`.
let mut download_config = build_download_config(
sync_mode,
Arc::new(rustc_hash::FxHashSet::default()),
Arc::clone(&asset_groupings),
Arc::from(lib_state.zone_name.as_str()),
);
// FINISHED is a prerequisite, not proof that this inventory remained
// complete. Deletion-capable maintenance stays deferred until the
// full scope and its final readiness check have qualified.
Arc::make_mut(&mut download_config).provisional =
indexing_reconciliation || !indexing_finished;
let managed_indexing_recovery = indexing_reconciliation
&& (download_config.edited_naming == crate::types::EditedNaming::Primary
|| match state_db {
Some(db) => db.has_primary_layouts(&lib_state.zone_name).await?,
None => false,
});
Arc::make_mut(&mut download_config).indexing_inventory_only =
managed_indexing_recovery && indexing_finished;
if pending_download_config_hash.is_some() && path_reconciliation_has_smart_folders {
Arc::make_mut(&mut download_config).complete_path_selection = true;
}
let download_client = shared_session.read().await.download_client().clone();
// Only an unrestricted current full-library selection can admit an exact
// sparse target without inventing album/rank membership.
let mut sparse_targets = Vec::new();
if download_controls.run_mode.downloads_files() && !is_retry_failed {
for target in library_states.iter().copied().filter(|target| {
target.zone_name != lib_state.zone_name
&& !target.plan_is_stale
&& target.plan.passes.len() == 1
}) {
let Some(pass) = target.plan.passes.first() else {
continue;
};
if pass.kind != crate::commands::PassKind::Unfiled || !pass.exclude_ids.is_empty() {
continue;
}
#[cfg(feature = "xmp")]
let target_groupings = if config.metadata.embed_xmp || config.metadata.xmp_sidecar {
preload_asset_groupings(state_db, &target.zone_name).await
} else {
Arc::new(download::AssetGroupings::default())
};
#[cfg(not(feature = "xmp"))]
let target_groupings = Arc::new(download::AssetGroupings::default());
let target_config = build_download_config(
download::SyncMode::Full,
Arc::clone(&pass.exclude_ids),
target_groupings,
Arc::from(target.zone_name.as_str()),
);
if target_config.recent.is_some()
|| target_config.skip_created_before.is_some()
|| target_config.skip_created_after.is_some()
|| target_config.retry_only
|| target_config.edited_naming == crate::types::EditedNaming::Primary
{
continue;
}
let mut target_config = target_config.with_pass(pass);
// A full-library path label is not evidence of album membership.
target_config.album_name = None;
sparse_targets.push(download::sparse_target::SelectedSparseTarget {
library: target.library.clone(),
album: pass.album.clone(),
config: Arc::new(target_config),
client: download_client.clone(),
controls: download_controls,
});
}
}
Arc::make_mut(&mut download_config).sparse_targets = sparse_targets.into();
let mut path_reconciliation_requires_smart_query = false;
if indexing_finished
&& !indexing_reconciliation
&& pending_download_config_hash.is_some()
&& download_controls.run_mode.downloads_files()
{
let reconciliation = download::reconcile_catalog_paths_with_client(
&download_client,
&lib_state.plan.passes,
Arc::clone(&download_config),
shutdown_token.clone(),
)
.await?;
path_reconciliation_requires_smart_query = lib_state
.plan
.passes
.iter()
.any(|pass| pass.kind == crate::commands::PassKind::SmartFolder);
path_reconciliation_complete &=
reconciliation.complete || reconciliation.complete_after_smart_query;
let reconciliation_failures = reconciliation
.stats
.failed
.saturating_add(reconciliation.stats.exif_failures)
.saturating_add(reconciliation.stats.state_write_failures);
cycle_failed_count = cycle_failed_count.saturating_add(reconciliation_failures);
cycle_stats.accumulate(&reconciliation.stats);
if reconciliation_failures > 0 {
// A normal download/adoption pass must not move state past a
// rejected reconciliation destination or failed finalization.
db_sync_token_advance_safe = false;
continue;
}
}
let mut sync_result = download::download_photos_with_sync(
&download_client,
&lib_state.plan.passes,
Arc::clone(&download_config),
download_controls,
shutdown_token.clone(),
)
.await?;
// Capture only the initial complete selected scope, before a bridge
// adds repeated count observations to the public activity statistics.
let inventory = sync_result.inventory.take().filter(|measurement| {
measurement.scope == download::inventory_scope(&lib_state.plan.passes, &download_config)
});
observe_sync_component(
&lib_state.zone_name,
"initial",
&sync_result,
lib_state.plan.passes.len(),
);
if indexing_finished
&& legacy_cycle.requires_inventory()
&& sync_result.full_enumeration_ran
&& matches!(
source_checkpoint_decision(
&sync_result,
config.runtime.dry_run,
lib_state.plan_is_stale,
CheckpointBasis::CompleteInventory
),
SourceCheckpointDecision::Advance { .. }
)
&& let Some(db) = state_db
{
let prior = db
.get_metadata(&lib_state.sync_token_key)
.await?
.unwrap_or_default();
if let Err(error) = legacy_cycle
.certify(
db,
&download_config,
&lib_state.library.all(),
preservation_config_hash.as_deref().unwrap_or(""),
&prior,
shutdown_token,
)
.await
{
download::legacy_preservation::log_preservation_hold("certification", &error);
}
}
let mut retained_checkpoint_expired = false;
let mut checkpoint_basis =
if sync_result.full_enumeration_ran && !sync_result.checkpoint.completed_delta_replay {
CheckpointBasis::CompleteInventory
} else {
CheckpointBasis::IncrementalDelta
};
// Eligibility reconciliation begins from a trusted prior cursor when
// one exists. The inventory covers unchanged newly eligible assets;
// replaying the prior cursor then bridges changes that occurred while
// that inventory was running. Only the bridged token may replace the
// active checkpoint.
let unresolved_identity = if let Some(db) = state_db {
db.get_metadata(&state::unresolved_identity_key(&lib_state.zone_name))
.await?
.is_some()
} else {
false
};
if (matches!(enum_config_hash_outcome, EnumConfigHashOutcome::Changed)
|| unresolved_identity
|| legacy_cycle.requires_inventory()
|| ((indexing_replay_pending || indexing_materialization_pending)
&& !is_retry_failed
&& !download_config.retry_only))
&& indexing_finished
&& !indexing_reconciliation
&& sync_result.full_enumeration_ran
&& !sync_result.checkpoint.completed_delta_replay
&& checkpoint_transition_state_safe
&& (!indexing_materialization_pending
|| (matches!(sync_result.outcome, download::DownloadOutcome::Success)
&& sync_result.stats.failed == 0
&& sync_result.stats.exif_failures == 0
&& download_config.recent.is_none()
&& download_config.skip_created_before.is_none()
&& download_config.skip_created_after.is_none()))
{
let prior_token = retained_cursor_at_start.clone();
if let Some(prior_token) = prior_token.filter(|token| !token.trim().is_empty()) {
let inventory_decision = source_checkpoint_decision(
&sync_result,
config.runtime.dry_run,
lib_state.plan_is_stale,
CheckpointBasis::CompleteInventory,
);
if matches!(inventory_decision, SourceCheckpointDecision::Advance { .. }) {
let mut bridge_config = build_download_config(
download::SyncMode::Incremental {
zone_sync_token: prior_token.clone(),
},
Arc::new(rustc_hash::FxHashSet::default()),
Arc::clone(&asset_groupings),
Arc::from(lib_state.zone_name.as_str()),
);
tracing::info!(
zone = %lib_state.zone_name,
"Inventory reconciliation complete; bridging changes from the preserved provider checkpoint"
);
Arc::make_mut(&mut bridge_config).sparse_targets =
Arc::clone(&download_config.sparse_targets);
let bridge_result = download::download_photos_with_sync(
&download_client,
&lib_state.plan.passes,
bridge_config,
download_controls,
shutdown_token.clone(),
)
.await?;
observe_sync_component(
&lib_state.zone_name,
"delta_bridge",
&bridge_result,
lib_state.plan.passes.len(),
);
let bridge_decision = source_checkpoint_decision(
&bridge_result,
config.runtime.dry_run,
lib_state.plan_is_stale,
CheckpointBasis::IncrementalDelta,
);
sync_result.outcome =
merge_download_outcomes(&sync_result.outcome, &bridge_result.outcome);
sync_result.accumulate(&bridge_result);
if bridge_result.checkpoint.retained_token_rejected
&& !bridge_result.checkpoint.completed_delta_replay
&& let Some(db) = state_db
{
retained_checkpoint_expired = true;
let mut hold = RetainedCheckpointHold::new(
&prior_token,
&enum_config_hash,
preservation_config_hash.as_deref(),
);
hold.attempt_count = retained_checkpoint_attempt;
if persist_retained_checkpoint_hold(
db,
retained_checkpoint_hold_key(&lib_state.zone_name),
&hold,
zone_metadata_fence.clone(),
)
.await
.is_err()
{
sync_result.checkpoint.state_write_failures = sync_result
.checkpoint
.state_write_failures
.saturating_add(1);
sync_result.stats.state_write_failures =
sync_result.stats.state_write_failures.saturating_add(1);
}
block_for_retained_checkpoint(&mut sync_result.stats);
}
if !bridge_result.full_enumeration_ran
|| bridge_result.checkpoint.completed_delta_replay
{
if let SourceCheckpointDecision::Advance { token, .. } = bridge_decision {
sync_result.sync_token = Some(token);
checkpoint_basis = CheckpointBasis::InventoryWithDeltaBridge;
} else {
sync_result.sync_token = None;
sync_result.checkpoint.sync_token_blocked = true;
sync_result.checkpoint.project(&mut sync_result.stats);
sync_result.stats.sync_token_blocked_reason =
Some("inventory_delta_bridge_failed");
sync_result.stats.sync_token_blocked_source = Some("kei");
sync_result.stats.sync_token_blocked_explanation = Some(
"the inventory completed, but replay from the preserved provider checkpoint did not finish safely",
);
}
} else {
sync_result.sync_token = None;
sync_result.checkpoint.sync_token_blocked = true;
sync_result.checkpoint.project(&mut sync_result.stats);
if bridge_result.checkpoint.retained_token_rejected {
block_for_retained_checkpoint(&mut sync_result.stats);
} else {
sync_result.stats.sync_token_blocked_reason =
Some("inventory_delta_bridge_failed");
sync_result.stats.sync_token_blocked_source = Some("kei");
sync_result.stats.sync_token_blocked_explanation = Some(
"the prior provider checkpoint could not be replayed incrementally after inventory reconciliation",
);
}
}
}
}
}
if unresolved_identity
&& !indexing_reconciliation
&& matches!(checkpoint_basis, CheckpointBasis::CompleteInventory)
{
// An inventory cannot prove that the previously ambiguous deltas
// were resolved. Require replay from the retained cursor.
sync_result.block_for_unresolved_identity();
}
if retained_checkpoint_expired {
block_for_retained_checkpoint(&mut sync_result.stats);
}
if sync_result.full_enumeration_ran && sync_result.stats.full_enumeration_reason.is_none() {
sync_result.stats.full_enumeration_reason = sync_mode_decision.full_enumeration_reason;
}
// --refresh-metadata bypasses automatic capture repair. Inspect the
// retained queue before inventory anchors or provider cursors commit,
// not just when reporting aggregate cycle health afterward.
if let Some(db) = state_db
&& let Some(capture) = db
.get_summary()
.await?
.metadata_capture
.into_iter()
.find(|capture| capture.library == lib_state.zone_name)
{
sync_result.stats.unattributed_legacy_assets = capture.unattributed_legacy_assets;
sync_result.stats.unattributed_legacy_pending = capture.unattributed_legacy_pending;
sync_result.stats.metadata_capture_remaining = capture.remaining_assets;
sync_result.stats.metadata_capture_unresolved = capture.unresolved_assets;
sync_result.stats.metadata_capture_deferred = capture.deferred_assets;
if capture.unresolved_assets > 0 {
sync_result.block_for_metadata_capture();
}
}
let library_completed_without_errors =
matches!(&sync_result.outcome, download::DownloadOutcome::Success)
&& !sync_result.checkpoint.identity_incomplete
&& !sync_result.checkpoint.interrupted
&& sync_result.checkpoint.enumeration_errors == 0
&& !shutdown_token.is_cancelled();
if path_reconciliation_requires_smart_query {
// Historical smart membership cannot certify the current selection.
// Full query proof is separate from a bounded source cursor hold.
path_reconciliation_complete &= library_completed_without_errors
&& download_controls.run_mode.downloads_files()
&& sync_result.full_enumeration_ran;
}
if should_warn_zero_assets(
&sync_result,
library_completed_without_errors,
download_controls.run_mode,
is_retry_failed,
lib_state,
) {
tracing::warn!(
library = %lib_state.zone_name,
library_count = library_states.len(),
assets_seen = sync_result.stats.assets_seen,
"Sync completed after enumerating zero assets; check iCloud library \
access and filters if this was unexpected"
);
}
if indexing_finished
&& !indexing_reconciliation
&& library_completed_without_errors
&& !sync_result.checkpoint.enumeration_incomplete
&& sync_result.checkpoint.state_write_failures == 0
&& !cycle_has_stale_plan
&& !is_retry_failed
&& download_controls.run_mode.downloads_files()
&& let Some(inventory) = inventory.as_ref()
&& let Some(db) = state_db
{
update_inventory_anchor_for_cycle(
db,
&enum_config_hash,
&lib_state.zone_name,
inventory,
&mut sync_result.stats,
)
.await;
}
// Readiness can regress during a long inventory. Recheck at the
// checkpoint boundary; even a failed probe leaves durable retry debt.
let readiness_after = tokio::select! {
biased;
() = shutdown_token.cancelled() => None,
result = async {
let mut finished = lib_state.library.check_indexing().await?;
for source in &lib_state.cross_zone_libraries {
if source.zone_name() != lib_state.zone_name {
finished &= source.check_indexing().await?;
}
}
Ok::<_, crate::icloud::error::ICloudError>(finished)
} => Some(result),
};
// Cancellation is not an indexing observation. Preserve existing debt,
// but do not turn ordinary interrupted sync into a new indexing hold.
let indexing_hold_required =
!indexing_finished || matches!(readiness_after, Some(Ok(false)) | Some(Err(_)));
if indexing_hold_required {
if !config.runtime.dry_run
&& let Some(db) = state_db
{
db.set_metadata(&indexing_key, "1").await?;
// This cycle must not stage or activate a token, even if all
// selected records happened to carry one unanimous candidate.
}
sync_result.checkpoint.sync_token_blocked = true;
sync_result.checkpoint.project(&mut sync_result.stats);
sync_result.stats.sync_token_blocked_reason = Some("indexing_reconciliation_required");
sync_result.stats.sync_token_blocked_source = Some("icloud");
sync_result.stats.sync_token_blocked_explanation = Some(
"indexing results are provisional; a FINISHED full reconciliation is required",
);
db_sync_token_advance_safe = false;
path_reconciliation_complete = false;
}
// Retry-only work cannot discharge either phase: new assets may be
// skipped before they acquire durable retry rows. Keep their cursor.
let recovery_retry_only = (indexing_reconciliation
|| indexing_replay_pending
|| indexing_materialization_pending)
&& (is_retry_failed || download_config.retry_only);
if recovery_retry_only
|| (indexing_reconciliation
&& (!sync_result.full_enumeration_ran
|| !library_completed_without_errors
|| sync_result.stats.failed > 0
|| sync_result.stats.exif_failures > 0))
{
sync_result.checkpoint.sync_token_blocked = true;
sync_result.checkpoint.project(&mut sync_result.stats);
if recovery_retry_only && !indexing_reconciliation {
sync_result.stats.sync_token_blocked_reason = Some("indexing_replay_required");
sync_result.stats.sync_token_blocked_source = Some("kei");
sync_result.stats.sync_token_blocked_explanation =
Some("retry-only processing cannot discharge retained-cursor replay");
}
}
// An inventory-only pass queued nothing. Unlike ordinary durable
// transfer retries, managed recovery cannot release its full-work debt
// until every selected publication and its retained bridge succeeds.
if indexing_materialization_pending
&& !indexing_reconciliation
&& (!library_completed_without_errors
|| !sync_result.full_enumeration_ran
|| sync_result.stats.failed > 0
|| sync_result.stats.exif_failures > 0
|| download_config.recent.is_some()
|| download_config.skip_created_before.is_some()
|| download_config.skip_created_after.is_some()
|| (retained_cursor_at_start
.as_ref()
.is_some_and(|s| !s.trim().is_empty())
&& !matches!(checkpoint_basis, CheckpointBasis::InventoryWithDeltaBridge)))
{
sync_result.checkpoint.sync_token_blocked = true;
sync_result.checkpoint.project(&mut sync_result.stats);
}
// Provider cursor safety is independent from transfer completion.
// The download pipeline persists every planned work item before this
// boundary, so failed transfers may advance while state, identity,
// enumeration, auth, and token-proof failures preserve the old cursor.
let mut checkpoint_decision =
if checkpoint_transition_state_safe && !shutdown_token.is_cancelled() {
source_checkpoint_decision(
&sync_result,
config.runtime.dry_run,
lib_state.plan_is_stale,
checkpoint_basis,
)
} else if shutdown_token.is_cancelled() {
SourceCheckpointDecision::Preserve {
reason: CheckpointHoldReason::Interrupted,
recovery: RecoveryAction::ReplayFromPriorToken,
}
} else {
SourceCheckpointDecision::Preserve {
reason: CheckpointHoldReason::StateNotDurable,
recovery: RecoveryAction::ReplayFromPriorToken,
}
};
if sync_result.stats.sync_token_blocked_reason == Some("retained_checkpoint_expired")
&& let SourceCheckpointDecision::Preserve { reason, .. } = checkpoint_decision
{
checkpoint_decision = SourceCheckpointDecision::Preserve {
reason,
recovery: RecoveryAction::AwaitRetainedCheckpointEvidence,
};
}
if indexing_reconciliation {
let mut qualified = false;
if matches!(
checkpoint_decision,
SourceCheckpointDecision::Advance { .. }
) && !is_retry_failed
&& !download_config.retry_only
&& !indexing_hold_required
&& indexing_finished
&& sync_result.full_enumeration_ran
&& library_completed_without_errors
&& (!managed_indexing_recovery
|| (matches!(readiness_after, Some(Ok(true)))
&& download_config.recent.is_none()
&& download_config.skip_created_before.is_none()
&& download_config.skip_created_after.is_none()))
&& let Some(db) = state_db
{
// Qualify coverage without adopting its cursor. Managed
// inventory queued no files, so retain separate full-work debt.
let mut obligations = vec![(replay_key.clone(), "1".to_owned())];
if managed_indexing_recovery {
obligations.push((materialization_key.clone(), "1".to_owned()));
}
match db
.commit_checkpoint_transition(state::CheckpointTransition {
expected_metadata: zone_metadata_fence.clone(),
legacy_preservation_proofs: Vec::new(),
legacy_config_hash: None,
sparse_identity_proofs: Vec::new(),
metadata_updates: obligations,
metadata_deletes: vec![indexing_key.clone()],
})
.await
{
Ok(()) => qualified = true,
Err(error) => {
tracing::warn!(error = %error, "Failed to qualify indexing inventory; retaining recovery hold");
sync_result.checkpoint.state_write_failures += 1;
sync_result.stats.state_write_failures += 1;
}
}
}
if matches!(
checkpoint_decision,
SourceCheckpointDecision::Advance { .. }
) {
checkpoint_decision = SourceCheckpointDecision::Preserve {
reason: CheckpointHoldReason::TokenProofIncomplete,
recovery: RecoveryAction::ReplayFromPriorToken,
};
}
sync_result.checkpoint.sync_token_blocked = true;
sync_result.checkpoint.project(&mut sync_result.stats);
sync_result.stats.sync_token_blocked_reason = Some(if qualified {
"indexing_inventory_qualified"
} else {
"indexing_reconciliation_required"
});
sync_result.stats.sync_token_blocked_source = Some("kei");
sync_result.stats.sync_token_blocked_explanation = Some(
if qualified && managed_indexing_recovery {
"managed inventory qualified; full protected materialization and retained-cursor replay are still required"
} else if qualified {
"full inventory qualified; the old trusted cursor is retained for the next incremental cycle"
} else {
"a complete non-retry inventory from a FINISHED plan is still required"
},
);
db_sync_token_advance_safe = false;
path_reconciliation_complete = false;
path_reconciliation_cycle_safe = false;
}
let mut legacy_proofs = Vec::new();
if let SourceCheckpointDecision::Advance { token, .. } = &checkpoint_decision
&& let Some(db) = state_db
{
let qualified = if legacy_cycle.requires_inventory()
&& (!matches!(sync_result.outcome, download::DownloadOutcome::Success)
|| sync_result.stats.failed > 0
|| sync_result.stats.exif_failures > 0)
{
Err(anyhow::anyhow!("Legacy current work is incomplete"))
} else {
legacy_cycle
.finish(db, &download_config, token, shutdown_token)
.await
};
match qualified {
Ok(proofs) => legacy_proofs = proofs,
Err(error) => {
download::legacy_preservation::log_preservation_hold("checkpoint", &error);
checkpoint_decision = SourceCheckpointDecision::Preserve {
reason: CheckpointHoldReason::LegacyPreservationIncomplete,
recovery: RecoveryAction::ReplayFromPriorToken,
};
sync_result.stats.sync_token_blocked = true;
sync_result.stats.sync_token_blocked_reason =
Some("legacy_preservation_incomplete");
sync_result.stats.sync_token_blocked_source = Some("kei");
sync_result.stats.sync_token_blocked_explanation = Some(
"legacy evidence is protected, but independent current coverage is incomplete",
);
}
}
}
path_reconciliation_cycle_safe &= library_completed_without_errors
&& checkpoint_transition_state_safe
&& !retained_checkpoint_expired
&& (matches!(
checkpoint_decision,
SourceCheckpointDecision::Advance { .. }
) || (matches!(
checkpoint_decision,
SourceCheckpointDecision::Preserve {
reason: CheckpointHoldReason::TokenProofIncomplete,
..
}
) && sync_result.checkpoint.bounded_path_selection_complete
&& !sync_result.checkpoint.enumeration_incomplete
&& sync_result.checkpoint.state_write_failures == 0
&& !legacy_cycle.requires_inventory()));
let mut checkpoint_evidence = checkpoint_support_fields(&checkpoint_decision);
let mut checkpoint_persistence = "no_state_db";
match checkpoint_decision {
SourceCheckpointDecision::Advance { token, basis } => {
crate::metrics::record_checkpoint_decision("advanced", basis.as_str());
crate::metrics::record_deferred_transfers(
sync_result
.stats
.failed
.saturating_add(sync_result.stats.exif_failures),
);
if let Some(db) = state_db {
let reconciliation_active = force_full_for_config_hash;
if reconciliation_active {
sparse_identity_proofs.extend(
sync_result
.checkpoint
.sparse_identity_proofs
.iter()
.cloned(),
);
let candidate_key =
pending_zone_token_key(&enum_config_hash, &lib_state.zone_name);
if let Err(e) = db
.commit_checkpoint_transition(state::CheckpointTransition {
expected_metadata: zone_metadata_fence.clone(),
legacy_preservation_proofs: legacy_proofs,
legacy_config_hash: preservation_config_hash.clone(),
sparse_identity_proofs: Vec::new(),
metadata_updates: vec![(candidate_key, token)],
metadata_deletes: vec![retained_checkpoint_hold_key(
&lib_state.zone_name,
)],
})
.await
{
checkpoint_hold_action =
Some(RecoveryAction::ReplayFromPriorToken.as_str());
db_sync_token_advance_safe = false;
path_reconciliation_cycle_safe = false;
tracing::warn!(
zone = %lib_state.zone_name,
error = %e,
"Failed to retain completed zone reconciliation checkpoint"
);
checkpoint_persistence = "failed";
} else {
checkpoint_persistence = "staged_reconciliation";
}
} else {
let mut metadata_updates = vec![(lib_state.sync_token_key.clone(), token)];
if let Some(recovery_action) = checkpoint_hold_action {
metadata_updates.push((
LAST_CHECKPOINT_STATUS_KEY.to_owned(),
"preserved".to_owned(),
));
metadata_updates.push((
LAST_RECOVERY_ACTION_KEY.to_owned(),
recovery_action.to_owned(),
));
} else {
metadata_updates.push((
LAST_CHECKPOINT_STATUS_KEY.to_owned(),
"current".to_owned(),
));
metadata_updates
.push((LAST_RECOVERY_ACTION_KEY.to_owned(), "none".to_owned()));
}
if let Err(e) = db
.commit_checkpoint_transition(state::CheckpointTransition {
expected_metadata: zone_metadata_fence.clone(),
legacy_preservation_proofs: legacy_proofs,
legacy_config_hash: preservation_config_hash.clone(),
sparse_identity_proofs: sync_result
.checkpoint
.sparse_identity_proofs
.clone(),
metadata_updates,
metadata_deletes: vec![
state::unresolved_identity_key(&lib_state.zone_name),
retained_checkpoint_hold_key(&lib_state.zone_name),
indexing_key.clone(),
replay_key.clone(),
materialization_key.clone(),
],
})
.await
{
checkpoint_hold_action =
Some(RecoveryAction::ReplayFromPriorToken.as_str());
db_sync_token_advance_safe = false;
path_reconciliation_cycle_safe = false;
tracing::warn!(error = %e, "Failed to store provider checkpoint");
checkpoint_persistence = "failed";
} else {
checkpoint_persistence = "stored";
if cycle_has_stale_plan && !lib_state.plan_is_stale {
tracing::warn!(
zone = %lib_state.zone_name,
diagnostic = "stale_plan_unaffected_zone",
"Stored clean zone checkpoint despite another selected zone's stale plan"
);
}
if sync_result.stats.failed > 0 || sync_result.stats.exif_failures > 0 {
tracing::info!(
zone = %lib_state.zone_name,
basis = ?basis,
deferred_transfers = sync_result.stats.failed,
metadata_failures = sync_result.stats.exif_failures,
"Provider checkpoint advanced: incomplete local work is durably queued for targeted retry"
);
} else {
tracing::debug!(zone = %lib_state.zone_name, basis = ?basis, "Stored provider checkpoint for next incremental sync");
}
}
}
} else {
db_sync_token_advance_safe = false;
tracing::debug!(
zone = %lib_state.zone_name,
"Provider checkpoint available but no state DB is configured"
);
}
}
SourceCheckpointDecision::Preserve { reason, recovery } => {
checkpoint_hold_action = Some(recovery.as_str());
crate::metrics::record_checkpoint_decision("preserved", reason.as_str());
db_sync_token_advance_safe = false;
if let Some(db) = state_db
&& let Err(e) = db
.commit_checkpoint_transition(state::CheckpointTransition {
expected_metadata: Vec::new(),
legacy_preservation_proofs: Vec::new(),
legacy_config_hash: None,
sparse_identity_proofs: Vec::new(),
metadata_updates: vec![
(
LAST_CHECKPOINT_STATUS_KEY.to_owned(),
"preserved".to_owned(),
),
(
LAST_RECOVERY_ACTION_KEY.to_owned(),
recovery.as_str().to_owned(),
),
],
metadata_deletes: Vec::new(),
})
.await
{
path_reconciliation_cycle_safe = false;
tracing::debug!(error = %e, "Failed to persist checkpoint hold status");
checkpoint_persistence = "failed";
} else if state_db.is_some() {
checkpoint_persistence = "stored_hold";
}
let diagnostic = sync_result
.stats
.sync_token_blocked_reason
.unwrap_or("provider_checkpoint_preserved");
let explanation = sync_result
.stats
.sync_token_blocked_explanation
.unwrap_or_else(|| download::sync_token_blocked_explanation(diagnostic));
tracing::warn!(
zone = %lib_state.zone_name,
reason = ?reason,
recovery = ?recovery,
diagnostic,
explanation,
"Provider checkpoint preserved; recovery will use the narrowest safe action"
);
}
}
if let Some(fields) = checkpoint_evidence.as_object_mut() {
fields.insert(
"persistence".into(),
serde_json::json!(checkpoint_persistence),
);
fields.insert(
"token_present".into(),
serde_json::json!(sync_result.sync_token.is_some()),
);
fields.insert(
"full_enumeration".into(),
serde_json::json!(sync_result.full_enumeration_ran),
);
fields.insert(
"identity_incomplete".into(),
serde_json::json!(sync_result.checkpoint.identity_incomplete),
);
fields.insert(
"state_write_failures".into(),
serde_json::json!(sync_result.checkpoint.state_write_failures),
);
fields.insert(
"enumeration_errors".into(),
serde_json::json!(sync_result.checkpoint.enumeration_errors),
);
if checkpoint_persistence == "failed" {
fields.insert(
"recovery".into(),
serde_json::json!("replay_from_prior_token"),
);
}
}
crate::support::observe_scoped(
&lib_state.zone_name,
None,
"support_checkpoint_v1",
checkpoint_evidence,
);
observe_sync_component(
&lib_state.zone_name,
"combined",
&sync_result,
lib_state.plan.passes.len(),
);
if sync_result.stats.sync_token_blocked
&& sync_result.stats.sync_token_blocked_zone.is_none()
{
sync_result.stats.sync_token_blocked_zone = Some(lib_state.zone_name.clone());
}
if sync_result.checkpoint.identity_incomplete {
cycle_failed_count += 1;
}
if sync_result.stats.unattributed_legacy_assets > 0
&& let Some(db) = state_db
&& let Some(capture) = db
.get_summary()
.await?
.metadata_capture
.into_iter()
.find(|capture| capture.library == lib_state.zone_name)
{
sync_result.stats.unattributed_legacy_assets = capture.unattributed_legacy_assets;
sync_result.stats.unattributed_legacy_pending = capture.unattributed_legacy_pending;
}
// Accumulate stats across libraries.
cycle_stats.accumulate(&sync_result.stats);
match sync_result.outcome {
download::DownloadOutcome::Success => {}
download::DownloadOutcome::SessionExpired { auth_error_count } => {
tracing::warn!(
auth_error_count,
zone = %lib_state.zone_name,
"Session expired during library sync"
);
cycle_session_expired = true;
break; // Stop iterating libraries -- need re-auth
}
download::DownloadOutcome::PartialFailure { failed_count } => {
cycle_failed_count += failed_count;
}
}
}
let reconciliation_active = force_full_for_config_hash;
if reconciliation_active
&& db_sync_token_advance_safe
&& !cycle_session_expired
&& !shutdown_token.is_cancelled()
&& let Some(db) = state_db
{
let mut metadata_updates = Vec::new();
let mut metadata_deletes = Vec::new();
if force_full_for_config_hash {
for lib_state in library_states
.iter()
.copied()
.filter(|state| has_active_passes(state))
{
let candidate_key = pending_zone_token_key(&enum_config_hash, &lib_state.zone_name);
let Some(token) = db
.get_metadata(&candidate_key)
.await?
.filter(|token| !token.trim().is_empty())
else {
db_sync_token_advance_safe = false;
path_reconciliation_cycle_safe = false;
tracing::warn!(
zone = %lib_state.zone_name,
"Config reconciliation has no completed checkpoint for selected zone"
);
break;
};
active_cursor_fences.push((candidate_key.clone(), Some(token.clone())));
metadata_updates.push((lib_state.sync_token_key.clone(), token));
metadata_deletes.push(candidate_key);
metadata_deletes.push(state::unresolved_identity_key(&lib_state.zone_name));
metadata_deletes.push(indexing_reconciliation_key(&lib_state.zone_name));
metadata_deletes.push(indexing_replay_key(&lib_state.zone_name));
metadata_deletes.push(indexing_materialization_key(&lib_state.zone_name));
}
if db_sync_token_advance_safe {
metadata_updates.push((ENUM_CONFIG_HASH_KEY.to_owned(), enum_config_hash));
metadata_deletes.push(PENDING_ENUM_CONFIG_HASH_KEY.to_owned());
}
}
if db_sync_token_advance_safe {
metadata_updates.push((LAST_CHECKPOINT_STATUS_KEY.to_owned(), "current".to_owned()));
metadata_updates.push((LAST_RECOVERY_ACTION_KEY.to_owned(), "none".to_owned()));
if let Err(e) = db
.commit_checkpoint_transition(state::CheckpointTransition {
expected_metadata: enum_metadata_at_start
.iter()
.cloned()
.chain(active_cursor_fences)
.collect(),
legacy_preservation_proofs: Vec::new(),
legacy_config_hash: preservation_config_hash.clone(),
sparse_identity_proofs,
metadata_updates,
metadata_deletes,
})
.await
{
db_sync_token_advance_safe = false;
path_reconciliation_cycle_safe = false;
tracing::warn!(error = %e, "Failed to atomically promote checkpoint reconciliation");
}
}
}
if path_reconciliation_complete
&& (!path_reconciliation_has_smart_folders
|| (cycle_failed_count == 0 && path_reconciliation_cycle_safe && !cycle_has_stale_plan))
&& !cycle_session_expired
&& !shutdown_token.is_cancelled()
&& let (Some(db), Some(download_config_hash)) = (state_db, pending_download_config_hash)
&& let Err(e) = db
.commit_checkpoint_transition(state::CheckpointTransition {
expected_metadata: path_metadata_at_start,
legacy_preservation_proofs: Vec::new(),
legacy_config_hash: None,
sparse_identity_proofs: Vec::new(),
metadata_updates: vec![(
download::DOWNLOAD_CONFIG_HASH_KEY.to_owned(),
download_config_hash,
)],
metadata_deletes: vec![PENDING_DOWNLOAD_CONFIG_HASH_KEY.to_owned()],
})
.await
{
tracing::warn!(error = %e, "Failed to promote completed local path reconciliation");
}
if let Some(db) = state_db {
let summary = db.get_summary().await?;
if summary.unresolved_identity_zones > 0 {
download::block_sync_token_for_unresolved_identity(&mut cycle_stats);
cycle_failed_count = cycle_failed_count.max(1);
db_sync_token_advance_safe = false;
}
if summary
.metadata_capture
.iter()
.any(|capture| capture.unattributed_legacy_pending > 0)
{
// Protected history no longer appears in the capture retry count,
// but pending current-work proof is still incomplete cycle work.
cycle_failed_count = cycle_failed_count.max(1);
db_sync_token_advance_safe = false;
}
if summary
.metadata_capture
.iter()
.any(|capture| capture.unresolved_assets > 0)
{
download::block_sync_token_for_metadata_capture(&mut cycle_stats);
cycle_failed_count = cycle_failed_count.max(1);
db_sync_token_advance_safe = false;
}
if summary
.last_recovery_action
.as_deref()
.is_some_and(|action| {
matches!(
action,
"await_retained_checkpoint_evidence"
| "retained_checkpoint_retry_exhausted"
| "repair_retained_checkpoint_evidence"
)
})
{
block_for_retained_checkpoint(&mut cycle_stats);
if summary.last_recovery_action.as_deref()
== Some("retained_checkpoint_retry_exhausted")
{
cycle_stats.sync_token_blocked_reason = Some("retained_checkpoint_retry_exhausted");
cycle_stats.sync_token_blocked_explanation = Some(
download::sync_token_blocked_explanation("retained_checkpoint_retry_exhausted"),
);
}
if summary.last_recovery_action.as_deref()
== Some("repair_retained_checkpoint_evidence")
{
cycle_stats.sync_token_blocked_reason = Some("retained_checkpoint_hold_invalid");
cycle_stats.sync_token_blocked_explanation = Some(
download::sync_token_blocked_explanation("retained_checkpoint_hold_invalid"),
);
}
cycle_failed_count = cycle_failed_count.max(1);
db_sync_token_advance_safe = false;
}
}
#[cfg(all(test, target_os = "linux"))]
crate::test_helpers::process_death_point("checkpoint-persisted");
Ok(CycleResult {
failed_count: cycle_failed_count,
session_expired: cycle_session_expired,
stats: cycle_stats,
db_sync_token_advance_safe,
})
}
/// Bulk-load `asset_albums` + `asset_people` into an in-memory index so the
/// filter phase can enrich payloads without per-asset DB hits. Scoped to a
/// single library so multi-library accounts don't cross-attribute album /
/// person memberships across zones (the v9 schema scopes both join tables
/// per library; this reader honours that scope).
#[cfg(feature = "xmp")]
pub(crate) async fn preload_asset_groupings<D>(
state_db: Option<&D>,
library: &str,
) -> Arc<download::AssetGroupings>
where
D: state::MembershipStore + ?Sized,
{
let Some(db) = state_db else {
return Arc::new(download::AssetGroupings::default());
};
let albums = db.get_all_asset_albums(library).await;
let people = db.get_all_asset_people(library).await;
let mut groupings = download::AssetGroupings::default();
match albums {
Ok(rows) => {
for (asset_id, album) in rows {
groupings.albums.entry(asset_id).or_default().push(album);
}
}
Err(e) => tracing::warn!(error = %e, library, "Failed to preload asset_albums"),
}
match people {
Ok(rows) => {
for (asset_id, person) in rows {
groupings.people.entry(asset_id).or_default().push(person);
}
}
Err(e) => tracing::warn!(error = %e, library, "Failed to preload asset_people"),
}
Arc::new(groupings)
}
/// Determine the sync mode for a library: full enumeration or incremental.
pub(crate) async fn determine_sync_mode_decision<D>(
_is_retry_failed: bool,
_library_count: usize,
state_db: Option<&D>,
sync_token_key: &str,
zone_name: &str,
) -> SyncModeDecision
where
D: state::SyncTokenStore + ?Sized,
{
if let Some(db) = state_db {
match db.get_metadata(sync_token_key).await {
Ok(Some(ref token)) if !token.is_empty() => {
tracing::debug!(zone = %zone_name, "Stored sync token found, using incremental sync");
SyncModeDecision {
mode: download::SyncMode::Incremental {
zone_sync_token: token.clone(),
},
full_enumeration_reason: None,
}
}
Ok(_) => {
let reason = download::FullEnumerationReason::NoStoredToken;
tracing::debug!(
zone = %zone_name,
full_enumeration_reason = reason.as_str(),
"No sync token found, performing full enumeration"
);
SyncModeDecision {
mode: download::SyncMode::Full,
full_enumeration_reason: Some(reason),
}
}
Err(e) => {
let reason = download::FullEnumerationReason::OtherStaticReason;
tracing::warn!(
error = %e,
full_enumeration_reason = reason.as_str(),
"Failed to load sync token, falling back to full enumeration"
);
SyncModeDecision {
mode: download::SyncMode::Full,
full_enumeration_reason: Some(reason),
}
}
}
} else {
SyncModeDecision {
mode: download::SyncMode::Full,
full_enumeration_reason: Some(download::FullEnumerationReason::NoStoredToken),
}
}
}
#[cfg(test)]
pub(crate) async fn determine_sync_mode<D>(
is_retry_failed: bool,
library_count: usize,
state_db: Option<&D>,
sync_token_key: &str,
zone_name: &str,
) -> download::SyncMode
where
D: state::SyncTokenStore + ?Sized,
{
determine_sync_mode_decision(
is_retry_failed,
library_count,
state_db,
sync_token_key,
zone_name,
)
.await
.mode
}
/// Keep strict-policy hashes byte-compatible with existing preservation proofs.
/// The opt-in affects only legacy evidence qualification, not download provenance.
pub(crate) fn hash_legacy_preservation_policy(
policy: &serde_json::Value,
allow_hardlinks: bool,
) -> String {
use sha2::{Digest, Sha256};
let mut hash = Sha256::new();
hash.update(policy.to_string().as_bytes());
if allow_hardlinks {
hash.update(b"\0legacy-preservation-allow-hardlinks:v1");
}
data_encoding::HEXLOWER.encode(&hash.finalize())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::commands::PassKind;
#[test]
fn support_component_totals_keep_inventory_and_bridge_separate() {
use tracing_subscriber::layer::SubscriberExt;
let (layer, drain) = crate::support::history::test_layer();
tracing::subscriber::with_default(tracing_subscriber::registry().with(layer), || {
let mut inventory = download::SyncResult::from_execution(
download::DownloadOutcome::Success,
Some("PRIVATE_TOKEN".into()),
download::SyncStats {
assets_seen: 3899,
api_total_at_start: Some(3899),
..Default::default()
},
);
inventory.full_enumeration_ran = true;
let bridge = download::SyncResult::from_execution(
download::DownloadOutcome::Success,
Some("PRIVATE_SUCCESSOR".into()),
download::SyncStats {
assets_seen: 238,
api_total_at_start: Some(238),
..Default::default()
},
);
observe_sync_component("PRIVATE_ZONE", "initial", &inventory, 1);
observe_sync_component("PRIVATE_ZONE", "delta_bridge", &bridge, 1);
inventory.accumulate(&bridge);
observe_sync_component("PRIVATE_ZONE", "combined", &inventory, 1);
});
let evidence = drain();
assert_eq!(evidence.len(), 3);
assert_eq!(evidence[0]["fields"]["assets_seen"], 3899);
assert_eq!(evidence[1]["fields"]["assets_seen"], 238);
assert_eq!(evidence[2]["fields"]["assets_seen"], 4137);
assert_eq!(evidence[0]["fields"]["component"], "initial");
assert!(evidence[2]["fields"]["inventory_complete"].is_null());
assert!(
!serde_json::to_string(&evidence)
.unwrap()
.contains("PRIVATE")
);
}
#[test]
fn retained_checkpoint_hold_round_trip_and_version_one_migration() {
let hold = RetainedCheckpointHold::new(
"private-cursor",
"enum-policy",
Some("preservation-policy"),
);
let encoded = serde_json::to_string(&hold).expect("encode receipt");
assert!(!encoded.contains("private-cursor"));
let decoded: RetainedCheckpointHold =
serde_json::from_str(&encoded).expect("decode receipt");
assert!(decoded.matches(&hold));
assert_eq!(decoded.attempt_count, 1);
assert_eq!(decoded.next_attempt_at, hold.next_attempt_at);
assert_eq!(serde_json::to_string(&decoded).unwrap(), encoded);
let mut version_one: serde_json::Value = serde_json::from_str(&encoded).unwrap();
version_one.as_object_mut().unwrap().remove("attempt_count");
let migrated: RetainedCheckpointHold = serde_json::from_value(version_one).unwrap();
assert!(migrated.matches(&hold));
assert_eq!(
migrated.attempt_count, 1,
"an existing receipt already consumed its first attempt"
);
let migrated_encoded = serde_json::to_string(&migrated).unwrap();
assert_eq!(
retained_checkpoint_hold_action(&migrated_encoded),
"await_retained_checkpoint_evidence"
);
let mut exhausted = migrated;
exhausted.attempt_count = RETAINED_CHECKPOINT_MAX_ATTEMPTS;
assert_eq!(
retained_checkpoint_hold_action(&serde_json::to_string(&exhausted).unwrap()),
"retained_checkpoint_retry_exhausted"
);
}
#[test]
fn legacy_preservation_trust_hash_preserves_strict_compatibility() {
let policy = serde_json::json!([1, 1, "coverage", "None", false, false, false]);
// Historical SHA-256 of the exact serialized strict policy, independently computed.
let strict = "6f307ea388dd8281b2db2d31c9a8ba37f3c910027bbdb47e95c014477ddf358f";
assert_eq!(hash_legacy_preservation_policy(&policy, false), strict);
assert_ne!(hash_legacy_preservation_policy(&policy, true), strict);
assert_eq!(hash_legacy_preservation_policy(&policy, false), strict);
}
fn make_library_state(has_passes: bool) -> LibraryState {
LibraryState {
library: crate::icloud::photos::PhotoLibrary::new_stub_with_zone(
Box::new(crate::test_helpers::MockPhotosSession::new()),
"PrimarySync",
),
cross_zone_libraries: Vec::new(),
pass_scope: crate::commands::PassScope {
include_albums: false,
include_smart_folders: false,
include_unfiled: has_passes,
},
zone_name: "PrimarySync".to_string(),
sync_token_key: "sync_token:PrimarySync".to_string(),
plan: crate::commands::AlbumPlan {
passes: if has_passes {
vec![crate::commands::AlbumPass {
kind: PassKind::Unfiled,
album: crate::icloud::photos::PhotoAlbum::stub_for_test(Arc::from(
"PrimarySync",
)),
exclude_ids: Arc::new(rustc_hash::FxHashSet::default()),
}]
} else {
Vec::new()
},
},
plan_is_stale: false,
plan_needs_refresh: false,
}
}
#[test]
fn should_warn_zero_assets_requires_active_passes() {
let sync_result = download::SyncResult {
full_enumeration_ran: true,
..download::SyncResult::from_execution(
download::DownloadOutcome::Success,
Some("zone-token".to_string()),
download::SyncStats {
assets_seen: 0,
..download::SyncStats::default()
},
)
};
assert!(!should_warn_zero_assets(
&sync_result,
true,
download::DownloadRunMode::Download,
false,
&make_library_state(false),
));
}
#[test]
fn should_warn_zero_assets_when_all_gates_are_true() {
let sync_result = download::SyncResult {
full_enumeration_ran: true,
..download::SyncResult::from_execution(
download::DownloadOutcome::Success,
Some("zone-token".to_string()),
download::SyncStats {
assets_seen: 0,
..download::SyncStats::default()
},
)
};
assert!(should_warn_zero_assets(
&sync_result,
true,
download::DownloadRunMode::Download,
false,
&make_library_state(true),
));
}
#[test]
fn durable_transfer_failure_can_advance_source_checkpoint() {
let result = download::SyncResult::from_execution(
download::DownloadOutcome::PartialFailure { failed_count: 1 },
Some("zone-token-next".to_string()),
download::SyncStats {
failed: 1,
..download::SyncStats::default()
},
);
assert_eq!(
source_checkpoint_decision(&result, false, false, CheckpointBasis::IncrementalDelta,),
SourceCheckpointDecision::Advance {
token: "zone-token-next".to_string(),
basis: CheckpointBasis::IncrementalDelta,
}
);
}
#[test]
fn metadata_capture_hold_survives_report_changes_and_result_composition() {
let mut result = download::SyncResult::from_execution(
download::DownloadOutcome::Success,
Some("zone-token-next".to_string()),
download::SyncStats::default(),
);
result.block_for_metadata_capture();
result.stats = download::SyncStats::default();
result.accumulate(&download::SyncResult::from_execution(
download::DownloadOutcome::Success,
None,
download::SyncStats::default(),
));
assert!(result.checkpoint.identity_incomplete);
assert!(result.stats.identity_incomplete);
assert_eq!(
source_checkpoint_decision(&result, false, false, CheckpointBasis::IncrementalDelta),
SourceCheckpointDecision::Preserve {
reason: CheckpointHoldReason::TokenProofIncomplete,
recovery: RecoveryAction::ReplayFromPriorToken,
}
);
}
#[test]
fn failed_retry_state_write_preserves_source_checkpoint() {
let result = download::SyncResult::from_execution(
download::DownloadOutcome::PartialFailure { failed_count: 1 },
Some("zone-token-next".to_string()),
download::SyncStats {
failed: 1,
state_write_failures: 1,
..download::SyncStats::default()
},
);
assert_eq!(
source_checkpoint_decision(&result, false, false, CheckpointBasis::IncrementalDelta,),
SourceCheckpointDecision::Preserve {
reason: CheckpointHoldReason::StateNotDurable,
recovery: RecoveryAction::ReplayFromPriorToken,
}
);
}
#[test]
fn incomplete_token_proof_preserves_concrete_retry_passes() {
let pass = download::PassKey {
index: 2,
kind: PassKind::Unfiled,
label: "unfiled".to_string(),
};
let mut result = download::SyncResult {
full_enumeration_ran: true,
..download::SyncResult::from_execution(
download::DownloadOutcome::Success,
None,
download::SyncStats {
sync_token_blocked: true,
..download::SyncStats::default()
},
)
};
result.checkpoint.retry_passes = vec![pass.clone()];
assert_eq!(
source_checkpoint_decision(&result, false, false, CheckpointBasis::CompleteInventory),
SourceCheckpointDecision::Preserve {
reason: CheckpointHoldReason::TokenProofIncomplete,
recovery: RecoveryAction::RetryPasses(vec![pass]),
}
);
}
#[test]
fn contract_source_checkpoint_requires_durable_recovery_exhaustive() {
let tokens = [Some("zone-token"), None, Some(" ")];
for mask in 0u16..1024 {
let dry_run = mask & 1 != 0;
let stale_pass_plan = mask & 2 != 0;
let session_expired = mask & 4 != 0;
let interrupted = mask & 8 != 0;
let enumeration_error = mask & 16 != 0;
let enumeration_incomplete = mask & 32 != 0;
let state_not_durable = mask & 64 != 0;
let token_proof_blocked = mask & 128 != 0;
let durable_transfer_failure = mask & 256 != 0;
let identity_incomplete = mask & 512 != 0;
let has_safety_blocker = mask & 255 != 0 || identity_incomplete;
for token in tokens {
let mut result = download::SyncResult::from_execution(
if session_expired {
download::DownloadOutcome::SessionExpired {
auth_error_count: 1,
}
} else if durable_transfer_failure {
download::DownloadOutcome::PartialFailure { failed_count: 1 }
} else {
download::DownloadOutcome::Success
},
token.map(str::to_string),
download::SyncStats {
interrupted,
enumeration_errors: usize::from(enumeration_error),
enumeration_incomplete,
state_write_failures: usize::from(state_not_durable),
sync_token_blocked: token_proof_blocked,
identity_incomplete,
..download::SyncStats::default()
},
);
let should_advance = !has_safety_blocker && token == Some("zone-token");
let decision = source_checkpoint_decision(
&result,
dry_run,
stale_pass_plan,
CheckpointBasis::IncrementalDelta,
);
assert_eq!(
matches!(decision, SourceCheckpointDecision::Advance { .. }),
should_advance,
"unexpected checkpoint decision for mask {mask:#010b} and token {token:?}"
);
// Reporting snapshots cannot remove or introduce safety evidence.
result.stats = download::SyncStats {
state_write_failures: usize::from(!state_not_durable),
enumeration_errors: usize::from(!enumeration_error),
enumeration_incomplete: !enumeration_incomplete,
interrupted: !interrupted,
sync_token_blocked: !token_proof_blocked,
identity_incomplete: !identity_incomplete,
..download::SyncStats::default()
};
let reported_decision = source_checkpoint_decision(
&result,
dry_run,
stale_pass_plan,
CheckpointBasis::IncrementalDelta,
);
assert_eq!(
reported_decision, decision,
"report mutation changed mask {mask:#010b}"
);
}
}
}
#[test]
fn inventory_drop_classifier_uses_five_percent_threshold() {
assert_eq!(classify_inventory_drop(100, 96), None);
assert_eq!(
classify_inventory_drop(100, 95),
Some(InventoryDrop {
previous_total: 100,
current_total: 95,
drop_assets: 5,
drop_percent: 5.0,
})
);
assert_eq!(classify_inventory_drop(0, 0), None);
assert_eq!(classify_inventory_drop(100, 101), None);
}
#[tokio::test]
async fn inventory_anchor_warns_and_updates_on_comparable_drop() {
let db = state::SqliteStateDb::open_in_memory().expect("state db");
let inventory = download::InventoryMeasurement {
scope: download::InventoryScope("test complete scope".to_owned()),
total: 80,
assets_seen: 80,
};
let key = inventory_anchor_key("hash-a", "PrimarySync", &inventory.scope);
let prior = InventoryAnchor {
version: INVENTORY_ANCHOR_VERSION,
scope: inventory.scope.clone(),
api_total_at_start: 100,
assets_seen: 100,
completed_at: 1_700_000_000,
};
db.set_metadata(&key, &serde_json::to_string(&prior).unwrap())
.await
.expect("seed anchor");
let mut stats = download::SyncStats {
api_total_at_start: Some(80),
assets_seen: 80,
..download::SyncStats::default()
};
update_inventory_anchor_for_cycle(&db, "hash-a", "PrimarySync", &inventory, &mut stats)
.await;
assert_eq!(stats.inventory_drop_warnings, 1);
assert_eq!(stats.inventory_drop_assets, 20);
assert_eq!(stats.inventory_drop_previous_total, Some(100));
assert_eq!(stats.inventory_drop_current_total, Some(80));
assert_eq!(
stats.inventory_drop_library,
Some("PrimarySync".to_string())
);
let raw = db
.get_metadata(&key)
.await
.expect("read anchor")
.expect("anchor exists");
let updated: InventoryAnchor = serde_json::from_str(&raw).expect("anchor json");
assert_eq!(updated.api_total_at_start, 80);
}
#[tokio::test]
async fn inventory_anchor_is_scoped_by_enum_config_hash() {
let db = state::SqliteStateDb::open_in_memory().expect("state db");
let inventory = download::InventoryMeasurement {
scope: download::InventoryScope("test complete scope".to_owned()),
total: 80,
assets_seen: 80,
};
let old_key = inventory_anchor_key("old-hash", "PrimarySync", &inventory.scope);
let prior = InventoryAnchor {
version: INVENTORY_ANCHOR_VERSION,
scope: inventory.scope.clone(),
api_total_at_start: 100,
assets_seen: 100,
completed_at: 1_700_000_000,
};
db.set_metadata(&old_key, &serde_json::to_string(&prior).unwrap())
.await
.expect("seed anchor");
let mut stats = download::SyncStats {
api_total_at_start: Some(80),
assets_seen: 80,
..download::SyncStats::default()
};
update_inventory_anchor_for_cycle(&db, "new-hash", "PrimarySync", &inventory, &mut stats)
.await;
assert_eq!(stats.inventory_drop_warnings, 0);
assert!(
db.get_metadata(&inventory_anchor_key(
"new-hash",
"PrimarySync",
&inventory.scope
))
.await
.expect("read new anchor")
.is_some(),
"new hash should get an independent anchor"
);
}
#[tokio::test]
async fn determine_sync_mode_two_normal_syncs_reuse_stored_token() {
let db = state::SqliteStateDb::open_in_memory().expect("state db");
let sync_token_key = sync_token_key("PrimarySync");
db.set_metadata(&sync_token_key, "stored-token-abc")
.await
.expect("set token");
for cycle in 1..=2 {
let mode =
determine_sync_mode(false, 1, Some(&db), &sync_token_key, "PrimarySync").await;
assert!(
matches!(mode, download::SyncMode::Incremental { ref zone_sync_token } if zone_sync_token == "stored-token-abc"),
"normal sync cycle {cycle} should use the stored token, got {mode:?}"
);
}
assert_eq!(
db.get_metadata(&sync_token_key).await.expect("read token"),
Some("stored-token-abc".to_string()),
"mode selection must not consume or clear the stored token"
);
}
#[test]
fn metadata_refresh_forces_full_enumeration() {
assert_eq!(
metadata_refresh_mode_decision(true),
Some(SyncModeDecision {
mode: download::SyncMode::Full,
full_enumeration_reason: Some(download::FullEnumerationReason::MetadataBackfill),
})
);
assert_eq!(metadata_refresh_mode_decision(false), None);
}
#[tokio::test]
async fn determine_sync_mode_decision_records_explicit_full_reasons() {
let db = state::SqliteStateDb::open_in_memory().expect("state db");
let sync_token_key = sync_token_key("PrimarySync");
db.set_metadata(&sync_token_key, "stored-token-abc")
.await
.expect("set token");
let retry =
determine_sync_mode_decision(true, 1, Some(&db), &sync_token_key, "PrimarySync").await;
assert!(matches!(
retry.mode,
download::SyncMode::Incremental { ref zone_sync_token }
if zone_sync_token == "stored-token-abc"
));
assert_eq!(retry.full_enumeration_reason, None);
db.delete_metadata_by_prefix(SYNC_TOKEN_PREFIX)
.await
.expect("clear token");
let no_token =
determine_sync_mode_decision(false, 1, Some(&db), &sync_token_key, "PrimarySync").await;
assert!(matches!(no_token.mode, download::SyncMode::Full));
assert_eq!(
no_token.full_enumeration_reason,
Some(download::FullEnumerationReason::NoStoredToken)
);
}
}