profit/lakehouse
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Phase 28-36 body of work

Browse Source 
Accumulated since a6f12e2 (Phase 21 Rust port + Phase 27 versioning):

- Phase 36: embed_semaphore on VectorState (permits=1) serializes
  seed embed calls — prevents sidecar socket collisions under
  concurrent /seed stress load
- Phase 31+: run_stress.ts 6-task diverse stress scaffolding;
  run_e2e_rated.ts + orchestrator.ts tightening
- Catalog dedupe cleanup: 16 duplicate manifests removed; canonical
  candidates.parquet (10.5MB -> 76KB) + placements.parquet (1.2MB ->
  11KB) regenerated post-dedupe; fresh manifests for active datasets
- vectord: harness EvalSet refinements (+181), agent portfolio
  rotation + ingest triggers (+158), autotune + rag adjustments
- catalogd/storaged/ingestd/mcp-server: misc tightening
- docs: Phase 28-36 PRD entries + DECISIONS ADR additions;
  control-plane pivot banner added to top of docs/PRD.md (pointing
  at docs/CONTROL_PLANE_PRD.md which lands in next commit)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
profit
2026-04-22 02:41:15 -05:00
co-authored by Claude Opus 4.7
parent a6f12e2609
commit 5b1fcf6d27
52 changed files with 2575 additions and 1012 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/catalogd/src/registry.rs
+452 -3
View File
@@ -28,6 +28,27 @@ pub struct MigrateBucketsReport {
pub manifests_persisted: usize,
}
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct DedupeReport {
pub groups: usize,
pub removed: usize,
pub kept: Vec<DedupeKept>,
pub errors: Vec<DedupeError>,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct DedupeKept {
pub name: String,
pub kept_id: String,
pub removed: usize,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct DedupeError {
pub manifest_id: String,
pub error: String,
}
/// Partial metadata update — only set fields are applied.
#[derive(Debug, Clone, Default, serde::Deserialize)]
pub struct MetadataUpdate {
@@ -212,13 +233,65 @@ impl Registry {
Ok(())
}
/// Register a new dataset. Persists manifest to storage before updating memory.
/// Register a dataset. Idempotent on `name`:
/// - No existing dataset by this name → create new manifest.
/// - Exists with same schema fingerprint → update objects + `updated_at`
/// in place (re-ingest of identical-shape data).
/// - Exists with different schema fingerprint → reject as `Err`
/// (callers map to HTTP 409 / gRPC FAILED_PRECONDITION). Schema
/// evolution is not yet exposed as an HTTP endpoint; callers must
/// register under a new name or delete the existing manifest first.
///
/// Concurrency: the write lock is held across the storage write. That
/// serializes concurrent registers but is correctness-mandatory —
/// dropping the lock across the check→insert sequence creates a TOCTOU
/// window where two callers with the same name both see "no existing"
/// and both insert, reintroducing the duplicate-manifest bug this
/// function exists to prevent. Register is a metadata-only hop on the
/// ingest path, so the serialization cost is acceptable. Precedent:
/// `update_metadata` also holds the write lock across its I/O.
///
/// Legacy state: if multiple manifests with the same name already exist
/// (from before this was idempotent), the most recently updated one is
/// treated as canonical. Run `dedupe_by_name` to clean up the rest.
pub async fn register(
&self,
name: String,
schema_fingerprint: SchemaFingerprint,
objects: Vec<ObjectRef>,
) -> Result<DatasetManifest, String> {
let mut datasets = self.datasets.write().await;
let existing = datasets
.values()
.filter(|d| d.name == name)
.max_by_key(|d| d.updated_at)
.cloned();
if let Some(mut manifest) = existing {
if manifest.schema_fingerprint != schema_fingerprint {
return Err(format!(
"dataset '{}' already exists with a different schema \
(existing fingerprint: {}, new: {}). Schema-evolution \
migration is not yet exposed as an HTTP endpoint; \
register under a new name or delete the existing \
manifest first.",
name, manifest.schema_fingerprint.0, schema_fingerprint.0,
));
}
manifest.objects = objects;
manifest.updated_at = chrono::Utc::now();
let manifest_key = format!("{MANIFEST_PREFIX}/{}.json", manifest.id);
let json = serde_json::to_vec_pretty(&manifest).map_err(|e| e.to_string())?;
ops::put(&self.store, &manifest_key, json.into()).await?;
datasets.insert(manifest.id.clone(), manifest.clone());
tracing::info!("re-registered (idempotent): {} ({})", manifest.name, manifest.id);
return Ok(manifest);
}
let now = chrono::Utc::now();
let manifest = DatasetManifest {
id: DatasetId::new(),
@@ -240,12 +313,10 @@ impl Registry {
embedding_refresh_policy: None,
};
// Write-ahead: persist before in-memory update
let manifest_key = format!("{MANIFEST_PREFIX}/{}.json", manifest.id);
let json = serde_json::to_vec_pretty(&manifest).map_err(|e| e.to_string())?;
ops::put(&self.store, &manifest_key, json.into()).await?;
let mut datasets = self.datasets.write().await;
datasets.insert(manifest.id.clone(), manifest.clone());
tracing::info!("registered dataset: {} ({})", manifest.name, manifest.id);
@@ -380,6 +451,72 @@ impl Registry {
Ok(manifest.clone())
}
/// Collapse duplicate manifests that share a `name`. Winner per group:
/// 1. Prefer a manifest with a non-null `row_count` (already resynced).
/// 2. Break ties by newest `updated_at`.
/// Losers are removed from the in-memory registry AND their manifest
/// JSON is deleted from object storage. Datasets with a single manifest
/// are untouched. Parquet data files are never touched — only catalog
/// metadata. See `register` for the prevention side of this fix.
///
/// Operator-only. Do not run while ingest is active: a concurrent
/// `register` between the snapshot and the delete sweep can create a
/// new manifest under a name we just deduped, producing a transient
/// count of 2 until the next dedupe run. Safe outside ingest windows.
pub async fn dedupe_by_name(&self) -> DedupeReport {
let groups: Vec<(String, Vec<DatasetManifest>)> = {
let datasets = self.datasets.read().await;
let mut by_name: HashMap<String, Vec<DatasetManifest>> = HashMap::new();
for m in datasets.values() {
by_name.entry(m.name.clone()).or_default().push(m.clone());
}
by_name.into_iter().filter(|(_, v)| v.len() > 1).collect()
};
let mut report = DedupeReport::default();
let mut to_delete: Vec<DatasetId> = Vec::new();
for (name, mut manifests) in groups {
report.groups += 1;
manifests.sort_by(|a, b| {
b.row_count.is_some().cmp(&a.row_count.is_some())
.then(b.updated_at.cmp(&a.updated_at))
});
let winner = &manifests[0];
report.kept.push(DedupeKept {
name: name.clone(),
kept_id: winner.id.to_string(),
removed: manifests.len() - 1,
});
for loser in &manifests[1..] {
to_delete.push(loser.id.clone());
}
}
{
let mut datasets = self.datasets.write().await;
for id in &to_delete {
datasets.remove(id);
}
}
for id in &to_delete {
let key = format!("{MANIFEST_PREFIX}/{}.json", id);
match ops::delete(&self.store, &key).await {
Ok(_) => report.removed += 1,
Err(e) => report.errors.push(DedupeError {
manifest_id: id.to_string(),
error: e,
}),
}
}
tracing::info!(
"dedupe: {} groups collapsed, {} manifests removed, {} errors",
report.groups, report.removed, report.errors.len(),
);
report
}
/// Resync every dataset that currently has a null row_count.
/// Returns (successes, failures) where each entry is (name, detail).
pub async fn resync_missing(&self) -> (Vec<(String, u64)>, Vec<(String, String)>) {
@@ -549,6 +686,46 @@ impl Registry {
Ok(())
}
/// Remove a dataset from the catalog by name. Deletes the manifest
/// from both the in-memory registry and object storage.
///
/// Scope is metadata-only: the underlying parquet files, vector
/// indexes, tombstones, trial journals, and AiViews that reference
/// this dataset are NOT touched. Caller is responsible for any
/// cascade cleanup. This mirrors how a DROP TABLE in a typical
/// warehouse separates "forget about this dataset" from "actually
/// reclaim the bytes".
///
/// Returns Ok(()) when the dataset existed and was removed; an Err
/// with "dataset not found" when no manifest by that name exists.
/// Legacy state with duplicate names: all matching manifests are
/// removed (effectively a post-hoc dedupe on that name).
pub async fn delete_dataset(&self, name: &str) -> Result<usize, String> {
let ids_to_remove: Vec<DatasetId> = {
let datasets = self.datasets.read().await;
datasets.values().filter(|d| d.name == name).map(|d| d.id.clone()).collect()
};
if ids_to_remove.is_empty() {
return Err(format!("dataset not found: {name}"));
}
for id in &ids_to_remove {
let key = format!("{MANIFEST_PREFIX}/{}.json", id);
if let Err(e) = ops::delete(&self.store, &key).await {
// Storage delete failed — log and keep going. The in-memory
// remove below is still correct; on restart, the missing
// storage object just means nothing to rehydrate.
tracing::warn!("delete_dataset '{name}': storage delete of {key} failed: {e}");
}
}
let mut datasets = self.datasets.write().await;
for id in &ids_to_remove {
datasets.remove(id);
}
tracing::info!("deleted dataset '{}' ({} manifest(s))", name, ids_to_remove.len());
Ok(ids_to_remove.len())
}
/// List datasets whose `embedding_stale_since` is set — they need a refresh.
pub async fn stale_datasets(&self) -> Vec<DatasetManifest> {
let datasets = self.datasets.read().await;
@@ -579,3 +756,275 @@ impl Registry {
Ok(manifest.clone())
}
}
#[cfg(test)]
mod tests {
use super::*;
use object_store::memory::InMemory;
fn fixture() -> Registry {
Registry::new(Arc::new(InMemory::new()))
}
fn fp(s: &str) -> SchemaFingerprint {
SchemaFingerprint(s.to_string())
}
fn obj(key: &str) -> ObjectRef {
ObjectRef {
bucket: "primary".to_string(),
key: key.to_string(),
size_bytes: 100,
created_at: chrono::Utc::now(),
}
}
#[tokio::test]
async fn fresh_register_creates_new_manifest() {
let reg = fixture();
let m = reg.register("a".into(), fp("f1"), vec![obj("a.parquet")]).await.unwrap();
assert_eq!(m.name, "a");
assert_eq!(m.schema_fingerprint, fp("f1"));
assert_eq!(reg.list().await.len(), 1);
assert_eq!(reg.get_by_name("a").await.unwrap().id, m.id);
}
#[tokio::test]
async fn re_register_same_fingerprint_is_idempotent() {
let reg = fixture();
let first = reg.register("a".into(), fp("f1"), vec![obj("old.parquet")]).await.unwrap();
let second = reg.register("a".into(), fp("f1"), vec![obj("new.parquet")]).await.unwrap();
assert_eq!(first.id, second.id, "same name+fp should reuse ID");
assert_eq!(reg.list().await.len(), 1, "no duplicate manifest created");
let fetched = reg.get_by_name("a").await.unwrap();
assert_eq!(fetched.objects[0].key, "new.parquet", "objects replaced");
assert!(fetched.updated_at >= first.updated_at, "updated_at bumped");
}
#[tokio::test]
async fn re_register_different_fingerprint_rejects() {
let reg = fixture();
reg.register("a".into(), fp("f1"), vec![obj("a.parquet")]).await.unwrap();
let err = reg.register("a".into(), fp("f2"), vec![obj("a.parquet")]).await.unwrap_err();
assert!(err.contains("different schema"), "error mentions schema drift: {err}");
assert_eq!(reg.list().await.len(), 1, "failed register did not mutate state");
assert_eq!(reg.get_by_name("a").await.unwrap().schema_fingerprint, fp("f1"));
}
#[tokio::test]
async fn delete_dataset_removes_manifest_and_in_memory_entry() {
let reg = fixture();
reg.register("to_delete".into(), fp("f1"), vec![obj("d.parquet")]).await.unwrap();
reg.register("keepme".into(), fp("f2"), vec![obj("k.parquet")]).await.unwrap();
assert_eq!(reg.list().await.len(), 2);
let removed = reg.delete_dataset("to_delete").await.unwrap();
assert_eq!(removed, 1, "one manifest removed");
assert!(reg.get_by_name("to_delete").await.is_none());
assert!(reg.get_by_name("keepme").await.is_some(), "unrelated dataset untouched");
assert_eq!(reg.list().await.len(), 1);
}
#[tokio::test]
async fn delete_dataset_404s_on_unknown_name() {
let reg = fixture();
let err = reg.delete_dataset("nope").await.unwrap_err();
assert!(err.starts_with("dataset not found"), "error wording: {err}");
}
#[tokio::test]
async fn delete_dataset_removes_all_dupes_with_matching_name() {
// Legacy zombies: direct insert so we can seed duplicates,
// mirroring pre-idempotent-register state. delete_dataset should
// sweep every manifest sharing that name in a single call.
let reg = fixture();
let now = chrono::Utc::now();
for _ in 0..3 {
let m = DatasetManifest {
id: DatasetId::new(),
name: "zombie".into(),
schema_fingerprint: fp("fp"),
objects: vec![obj("z.parquet")],
created_at: now, updated_at: now,
row_count: None, description: String::new(), owner: String::new(),
sensitivity: None, columns: vec![], lineage: None, freshness: None,
tags: vec![], last_embedded_at: None, embedding_stale_since: None,
embedding_refresh_policy: None,
};
reg.datasets.write().await.insert(m.id.clone(), m);
}
assert_eq!(reg.list().await.len(), 3);
let removed = reg.delete_dataset("zombie").await.unwrap();
assert_eq!(removed, 3, "all three manifests removed in one sweep");
assert_eq!(reg.list().await.len(), 0);
}
#[tokio::test]
async fn dedupe_no_duplicates_is_noop() {
let reg = fixture();
reg.register("a".into(), fp("f1"), vec![obj("a.parquet")]).await.unwrap();
reg.register("b".into(), fp("f2"), vec![obj("b.parquet")]).await.unwrap();
let report = reg.dedupe_by_name().await;
assert_eq!(report.groups, 0);
assert_eq!(report.removed, 0);
assert_eq!(reg.list().await.len(), 2);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn concurrent_register_stress_many_workers_one_manifest() {
// Stress variant: fire 32 concurrent registers for the same new
// name across 8 worker threads. Single-manifest invariant must
// hold regardless of how the scheduler interleaves them. This
// catches race regressions the 2-call variant below might miss
// under a forgiving scheduler.
let reg = fixture();
let mut handles = Vec::new();
for i in 0..32 {
let r = reg.clone();
handles.push(tokio::spawn(async move {
r.register(
"stress".into(),
fp("stress-fp"),
vec![obj(&format!("{i}.parquet"))],
).await
}));
}
for h in handles {
h.await.unwrap().expect("each register succeeds (idempotent)");
}
let all = reg.list().await;
let stress_manifests: Vec<_> = all.iter().filter(|d| d.name == "stress").collect();
assert_eq!(stress_manifests.len(), 1,
"32 concurrent registers produced {} manifests for 'stress' — expected 1",
stress_manifests.len());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_register_same_new_name_collapses_to_one_manifest() {
// Pins the TOCTOU fix: without the write-lock held across the
// check→insert sequence, two parallel registers of the same
// previously-unknown name would both see "no existing" and both
// insert. Multi-thread flavor is required to actually exercise
// the race — under single-threaded tokio the scheduler doesn't
// preempt between awaits on the same task's critical section.
// Under the current impl, whichever acquires the write lock
// first creates the manifest; the second observes it and takes
// the idempotent-update branch.
let reg = fixture();
let a = reg.clone();
let b = reg.clone();
let (ra, rb) = tokio::join!(
async move {
a.register("race".into(), fp("shared-fp"), vec![obj("a.parquet")]).await
},
async move {
b.register("race".into(), fp("shared-fp"), vec![obj("b.parquet")]).await
},
);
let m_a = ra.expect("first register succeeds");
let m_b = rb.expect("second register succeeds (same fp = idempotent)");
assert_eq!(m_a.id, m_b.id, "both calls resolve to a single DatasetId");
assert_eq!(reg.list().await.len(), 1, "no duplicate manifest created");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_register_same_name_different_fp_one_wins_one_rejects() {
// Reverse case: concurrent registers with conflicting fingerprints.
// Deterministic outcome: one call's fingerprint gets persisted as
// canonical, the other is rejected on seeing the now-established
// fingerprint. Which one wins is scheduler-dependent — we only
// assert exactly one success and exactly one rejection.
let reg = fixture();
let a = reg.clone();
let b = reg.clone();
let (ra, rb) = tokio::join!(
async move { a.register("race".into(), fp("fp1"), vec![obj("a.parquet")]).await },
async move { b.register("race".into(), fp("fp2"), vec![obj("b.parquet")]).await },
);
let successes = [&ra, &rb].iter().filter(|r| r.is_ok()).count();
let rejections = [&ra, &rb].iter().filter(|r| r.is_err()).count();
assert_eq!(successes, 1, "exactly one register wins");
assert_eq!(rejections, 1, "the other is rejected on fingerprint");
let rejection_msg = [&ra, &rb].iter()
.find_map(|r| r.as_ref().err())
.unwrap();
assert!(rejection_msg.contains("different schema"));
assert_eq!(reg.list().await.len(), 1);
}
#[tokio::test]
async fn dedupe_collapses_dupes_preferring_non_null_row_count() {
let reg = fixture();
// Simulate legacy pre-idempotent-register state by inserting
// manifests with the same `name` directly. Mirrors what happened
// with the 308× successful_playbooks duplication.
let now = chrono::Utc::now();
let old_with_rows = DatasetManifest {
id: DatasetId::new(),
name: "dupes".into(),
schema_fingerprint: fp("f1"),
objects: vec![obj("dupes.parquet")],
created_at: now - chrono::Duration::hours(2),
updated_at: now - chrono::Duration::hours(2),
row_count: Some(42),
description: String::new(),
owner: String::new(),
sensitivity: None,
columns: vec![],
lineage: None,
freshness: None,
tags: vec![],
last_embedded_at: None,
embedding_stale_since: None,
embedding_refresh_policy: None,
};
let newer_no_rows = DatasetManifest {
id: DatasetId::new(),
updated_at: now, // newer, but null row_count
row_count: None,
..old_with_rows.clone()
};
let oldest = DatasetManifest {
id: DatasetId::new(),
updated_at: now - chrono::Duration::hours(5),
row_count: None,
..old_with_rows.clone()
};
let kept_id = old_with_rows.id.clone();
let loser_a = newer_no_rows.id.clone();
let loser_b = oldest.id.clone();
{
let mut d = reg.datasets.write().await;
d.insert(old_with_rows.id.clone(), old_with_rows);
d.insert(newer_no_rows.id.clone(), newer_no_rows);
d.insert(oldest.id.clone(), oldest);
}
assert_eq!(reg.list().await.len(), 3);
let report = reg.dedupe_by_name().await;
assert_eq!(report.groups, 1);
assert_eq!(report.removed, 2);
assert_eq!(report.errors.len(), 0);
assert_eq!(report.kept.len(), 1);
assert_eq!(report.kept[0].name, "dupes");
assert_eq!(report.kept[0].kept_id, kept_id.to_string(),
"should keep the manifest with non-null row_count");
let remaining = reg.list().await;
assert_eq!(remaining.len(), 1);
assert_eq!(remaining[0].id, kept_id);
assert!(reg.get(&loser_a).await.is_none());
assert!(reg.get(&loser_b).await.is_none());
}
}