Handle Notion database roots returned as type mismatches

crates/locality-engine/src/synchronize_project.rs

@@ -9,8 +9,8 @@ use std::collections::{BTreeMap, BTreeSet};
 use locality_connector::{
     Connector, PortableArtifactKey, PortableBootstrapRequest, PortableChangeBatch,
     PortableCompleteness, PortableContentArtifact, PortableEnumerateRequest,
-    PortableEnumerateResult, PortableFetchReason, PortableFetchRequest, PortableProjectionArtifact,
-    PortableRenderRequest, PortableSourceChange, PortableSyncRequest,
+    PortableEnumerateResult, PortableFetchReason, PortableFetchRequest, PortableIncompleteReason,
+    PortableProjectionArtifact, PortableRenderRequest, PortableSourceChange, PortableSyncRequest,
     portable_scope_root_remote_id,
 };
 use locality_core::model::RemoteId;
@@ -93,6 +93,32 @@ impl UnpersistedSynchronizationBatch {
     }
 }
 
+/// Hard bounds for aggregating a paginated portable bootstrap.
+///
+/// These limits apply to the aggregate rather than to an individual connector
+/// request. `PortableBootstrapRequest::max_changes` remains the per-checkpoint
+/// provider-work bound.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct BootstrapAggregationLimits {
+    pub max_checkpoints: usize,
+    pub max_total_changes: usize,
+    pub max_total_content_bytes: u64,
+}
+
+impl BootstrapAggregationLimits {
+    fn validate(self) -> LocalityResult<Self> {
+        if self.max_checkpoints == 0
+            || self.max_total_changes == 0
+            || self.max_total_content_bytes == 0
+        {
+            return Err(aggregation_error(
+                "portable bootstrap aggregation limits must be nonzero",
+            ));
+        }
+        Ok(self)
+    }
+}
+
 /// Run one bootstrap checkpoint through fetch and render.
 pub fn bootstrap_and_project<C: Connector + ?Sized>(
     connector: &C,
@@ -110,6 +136,271 @@ pub fn bootstrap_and_project<C: Connector + ?Sized>(
     )
 }
 
+/// Run every checkpoint of one bounded bootstrap and return one deterministic
+/// unpersisted candidate.
+///
+/// `CheckpointContinuation` is pagination control flow, not a coverage gap.
+/// Every other incomplete reason is retained and therefore continues to block
+/// publication after the terminal checkpoint is reached.
+pub fn bootstrap_and_project_to_completion<C: Connector + ?Sized>(
+    connector: &C,
+    request: PortableBootstrapRequest,
+    format_version: u32,
+    limits: BootstrapAggregationLimits,
+) -> LocalityResult<UnpersistedSynchronizationBatch> {
+    let limits = limits.validate()?;
+    let expected_source_connection_id = request.source_connection_id.clone();
+    let scope = request.scope;
+    let max_changes = request.max_changes;
+    let mut current_checkpoint = request.checkpoint;
+    let mut seen_checkpoints = BTreeSet::new();
+    if let Some(checkpoint) = &current_checkpoint {
+        seen_checkpoints.insert(checkpoint_identity(checkpoint));
+    }
+    let mut aggregate = BootstrapAggregate::new(expected_source_connection_id.clone());
+    let mut checkpoint_count = 0_usize;
+
+    loop {
+        if checkpoint_count >= limits.max_checkpoints {
+            return Err(aggregation_error(
+                "portable bootstrap aggregation exceeded its checkpoint limit",
+            ));
+        }
+        checkpoint_count += 1;
+        let page = bootstrap_and_project(
+            connector,
+            PortableBootstrapRequest {
+                source_connection_id: expected_source_connection_id.clone(),
+                scope: scope.clone(),
+                checkpoint: current_checkpoint.clone(),
+                max_changes,
+            },
+            format_version,
+        )
+        .map_err(|_| aggregation_error("portable bootstrap aggregation page failed"))?;
+        if page.source_connection_id != expected_source_connection_id {
+            return Err(aggregation_error(
+                "portable bootstrap aggregation changed source connection",
+            ));
+        }
+
+        let (continuation, preserved_completeness) =
+            completeness_without_continuation(&page.completeness);
+        if continuation {
+            validate_continuation_checkpoint(
+                current_checkpoint.as_ref(),
+                &page.next_checkpoint,
+                &mut seen_checkpoints,
+            )?;
+        }
+        let next_checkpoint = page.next_checkpoint.clone();
+        aggregate.push(page, preserved_completeness, limits)?;
+
+        if !continuation {
+            return Ok(aggregate.finish(next_checkpoint));
+        }
+        current_checkpoint = Some(next_checkpoint);
+    }
+}
+
+struct BootstrapAggregate {
+    source_connection_id: SourceConnectionId,
+    observed_changes: BTreeMap<RemoteId, PortableSourceChange>,
+    source_versions: BTreeMap<RemoteId, ImmutableSourceVersionCandidate>,
+    contents: BTreeMap<PortableArtifactKey, ImmutableContentCandidate>,
+    projections: BTreeMap<PortableArtifactKey, ImmutableProjectionCandidate>,
+    projection_paths: BTreeSet<String>,
+    completeness: PortableCompleteness,
+    total_changes: usize,
+    total_content_bytes: u64,
+}
+
+impl BootstrapAggregate {
+    fn new(source_connection_id: SourceConnectionId) -> Self {
+        Self {
+            source_connection_id,
+            observed_changes: BTreeMap::new(),
+            source_versions: BTreeMap::new(),
+            contents: BTreeMap::new(),
+            projections: BTreeMap::new(),
+            projection_paths: BTreeSet::new(),
+            completeness: PortableCompleteness::complete(),
+            total_changes: 0,
+            total_content_bytes: 0,
+        }
+    }
+
+    fn push(
+        &mut self,
+        page: UnpersistedSynchronizationBatch,
+        preserved_completeness: PortableCompleteness,
+        limits: BootstrapAggregationLimits,
+    ) -> LocalityResult<()> {
+        for source in &page.source_versions {
+            if self
+                .source_versions
+                .contains_key(&source.source_object.remote_id)
+            {
+                return Err(aggregation_error(
+                    "portable bootstrap aggregation repeated a source version",
+                ));
+            }
+        }
+        for change in &page.observed_changes {
+            if self
+                .observed_changes
+                .contains_key(&change.source_object.remote_id)
+            {
+                return Err(aggregation_error(
+                    "portable bootstrap aggregation repeated an observed source",
+                ));
+            }
+        }
+        for projection in &page.projections {
+            if self.projections.contains_key(&projection.artifact_key) {
+                return Err(aggregation_error(
+                    "portable bootstrap aggregation repeated a projection artifact",
+                ));
+            }
+            if self
+                .projection_paths
+                .contains(projection.logical_path.as_str())
+            {
+                return Err(aggregation_error(
+                    "portable bootstrap aggregation repeated a logical path",
+                ));
+            }
+        }
+        for content in &page.contents {
+            if self.contents.contains_key(&content.artifact_key) {
+                return Err(aggregation_error(
+                    "portable bootstrap aggregation repeated a content artifact",
+                ));
+            }
+        }
+
+        let total_changes = self
+            .total_changes
+            .checked_add(page.observed_changes.len())
+            .ok_or_else(|| {
+                aggregation_error("portable bootstrap aggregation change count overflowed")
+            })?;
+        if total_changes > limits.max_total_changes {
+            return Err(aggregation_error(
+                "portable bootstrap aggregation exceeded its change limit",
+            ));
+        }
+        let page_content_bytes = page.contents.iter().try_fold(0_u64, |total, content| {
+            total.checked_add(content.byte_length).ok_or_else(|| {
+                aggregation_error("portable bootstrap aggregation content bytes overflowed")
+            })
+        })?;
+        let total_content_bytes = self
+            .total_content_bytes
+            .checked_add(page_content_bytes)
+            .ok_or_else(|| {
+                aggregation_error("portable bootstrap aggregation content bytes overflowed")
+            })?;
+        if total_content_bytes > limits.max_total_content_bytes {
+            return Err(aggregation_error(
+                "portable bootstrap aggregation exceeded its content byte limit",
+            ));
+        }
+
+        self.total_changes = total_changes;
+        self.total_content_bytes = total_content_bytes;
+        self.completeness.merge(preserved_completeness);
+        self.source_versions.extend(
+            page.source_versions
+                .into_iter()
+                .map(|source| (source.source_object.remote_id.clone(), source)),
+        );
+        self.observed_changes.extend(
+            page.observed_changes
+                .into_iter()
+                .map(|change| (change.source_object.remote_id.clone(), change)),
+        );
+        self.contents.extend(
+            page.contents
+                .into_iter()
+                .map(|content| (content.artifact_key.clone(), content)),
+        );
+        for projection in page.projections {
+            self.projection_paths
+                .insert(projection.logical_path.as_str().to_string());
+            self.projections
+                .insert(projection.artifact_key.clone(), projection);
+        }
+        Ok(())
+    }
+
+    fn finish(
+        self,
+        next_checkpoint: locality_connector::PortableCheckpoint,
+    ) -> UnpersistedSynchronizationBatch {
+        let publication_eligible = self.completeness.is_complete();
+        UnpersistedSynchronizationBatch {
+            source_connection_id: self.source_connection_id,
+            observed_changes: self.observed_changes.into_values().collect(),
+            source_versions: self.source_versions.into_values().collect(),
+            contents: self.contents.into_values().collect(),
+            projections: self.projections.into_values().collect(),
+            next_checkpoint,
+            completeness: self.completeness,
+            publication_eligible,
+        }
+    }
+}
+
+fn completeness_without_continuation(
+    completeness: &PortableCompleteness,
+) -> (bool, PortableCompleteness) {
+    let reasons = completeness.incomplete_reasons();
+    let continuation = reasons.contains(&PortableIncompleteReason::CheckpointContinuation);
+    let mut preserved = if reasons.is_empty() && !completeness.is_complete() {
+        PortableCompleteness::default()
+    } else {
+        PortableCompleteness::complete()
+    };
+    for reason in reasons {
+        if reason != &PortableIncompleteReason::CheckpointContinuation {
+            preserved.merge(PortableCompleteness::incomplete(reason.clone()));
+        }
+    }
+    (continuation, preserved)
+}
+
+fn validate_continuation_checkpoint(
+    current: Option<&locality_connector::PortableCheckpoint>,
+    next: &locality_connector::PortableCheckpoint,
+    seen: &mut BTreeSet<(u16, String)>,
+) -> LocalityResult<()> {
+    if next.opaque.is_empty() {
+        return Err(aggregation_error(
+            "portable bootstrap continuation returned an empty checkpoint",
+        ));
+    }
+    if current == Some(next) {
+        return Err(aggregation_error(
+            "portable bootstrap continuation repeated its checkpoint",
+        ));
+    }
+    if !seen.insert(checkpoint_identity(next)) {
+        return Err(aggregation_error(
+            "portable bootstrap continuation formed a checkpoint cycle",
+        ));
+    }
+    Ok(())
+}
+
+fn checkpoint_identity(checkpoint: &locality_connector::PortableCheckpoint) -> (u16, String) {
+    (checkpoint.format_version, checkpoint.opaque.clone())
+}
+
+fn aggregation_error(message: &'static str) -> LocalityError {
+    LocalityError::InvalidState(message.to_string())
+}
+
 /// Run one incremental synchronization checkpoint through fetch and render.
 pub fn synchronize_and_project_portable<C: Connector + ?Sized>(
     connector: &C,