From 3101789d32f34630da7a70684b468857842e882a Mon Sep 17 00:00:00 2001 From: Alan Kahn Date: Tue, 3 Mar 2026 09:57:19 -0500 Subject: [PATCH 1/8] feat: k8s health lifecycle + standalone Dockerfile for internal deployment Add /health/startup and /health/live probes, flip readiness to 503 on SIGTERM so k8s stops routing before drain. Update standalone Dockerfile for internal deployment: port 9500, schema mounted at runtime (not baked in), persistent data dir, non-root user with fixed UID 10001. Co-Authored-By: Claude Opus 4.6 --- docker/standalone/Dockerfile | 34 +++++++++++++--------------- tidal-server/src/main.rs | 13 +++++++---- tidal-server/src/router.rs | 43 +++++++++++++++++++++++++++--------- tidal-server/src/state.rs | 15 ++++++++++++- 4 files changed, 70 insertions(+), 35 deletions(-) diff --git a/docker/standalone/Dockerfile b/docker/standalone/Dockerfile index 8c84fd3..44ca6fe 100644 --- a/docker/standalone/Dockerfile +++ b/docker/standalone/Dockerfile @@ -1,5 +1,6 @@ -FROM rust:1.91 AS builder -WORKDIR /app +FROM rust:1.91-slim AS builder +WORKDIR /build +RUN apt-get update && apt-get install -y pkg-config libssl-dev && rm -rf /var/lib/apt/lists/* # Copy workspace manifests first for layer caching. COPY Cargo.toml Cargo.lock ./ @@ -16,20 +17,15 @@ COPY . . RUN cargo build -p tidal-server --release FROM debian:bookworm-slim -WORKDIR /srv -RUN useradd --system --home /srv tidal && \ - apt-get update && apt-get install -y ca-certificates curl && \ - rm -rf /var/lib/apt/lists/* - -COPY --from=builder /app/target/release/tidal-server /usr/local/bin/tidal-server -COPY tidal-server/config /etc/tidal-server - -USER tidal -EXPOSE 9400 9091 - -HEALTHCHECK --interval=30s --timeout=5s --start-period=15s --retries=3 \ - CMD curl -f -H "Authorization: Bearer ${TIDAL_API_KEY:-}" http://localhost:9400/health || exit 1 - -ENTRYPOINT ["tidal-server", "standalone", \ - "--listen", "0.0.0.0:9400", \ - "--metrics", "0.0.0.0:9091"] +RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates curl && rm -rf /var/lib/apt/lists/* +RUN useradd -m -u 10001 tidal +COPY --from=builder --chown=tidal:tidal /build/target/release/tidal-server /usr/local/bin/tidal-server +RUN mkdir -p /config && chown tidal:tidal /config +USER tidal:tidal +WORKDIR /data +EXPOSE 9500 +HEALTHCHECK --interval=30s --timeout=3s --start-period=10s --retries=3 \ + CMD curl -sf http://localhost:9500/health || exit 1 +ENV TIDAL_SERVER_LOG=info +ENTRYPOINT ["/usr/local/bin/tidal-server"] +CMD ["standalone", "--listen", "0.0.0.0:9500", "--schema", "/config/eros-schema.yaml", "--data-dir", "/data"] diff --git a/tidal-server/src/main.rs b/tidal-server/src/main.rs index 627b1dc..f50be46 100644 --- a/tidal-server/src/main.rs +++ b/tidal-server/src/main.rs @@ -111,13 +111,16 @@ async fn serve(state: ServerState, addr: &str, api_key: Option>) -> Res let actual = listener.local_addr()?; tracing::info!("listening on http://{actual}"); - axum::serve(listener, build_router(Arc::new(state), api_key)) - .with_graceful_shutdown(shutdown_signal()) + let state = Arc::new(state); + let shutdown_state = state.clone(); + + axum::serve(listener, build_router(state, api_key)) + .with_graceful_shutdown(shutdown_signal(shutdown_state)) .await?; Ok(()) } -async fn shutdown_signal() { +async fn shutdown_signal(state: Arc) { // SIGTERM is Unix-only; on other platforms we fall back to ctrl-c alone. #[cfg(unix)] let sigterm = async { @@ -146,5 +149,7 @@ async fn shutdown_signal() { _ = sigterm => {} } - tracing::info!("shutdown signal received"); + // Flip readiness BEFORE axum starts draining + state.set_shutting_down(); + tracing::info!("readiness flipped to not-ready, draining in-flight requests"); } diff --git a/tidal-server/src/router.rs b/tidal-server/src/router.rs index 827663a..fee1eb8 100644 --- a/tidal-server/src/router.rs +++ b/tidal-server/src/router.rs @@ -75,6 +75,8 @@ pub fn build_router(state: Arc, api_key: Option>) -> Route // Public routes — exempt from auth so health probes always work. let public = Router::new() .route("/health", get(health)) + .route("/health/startup", get(health_startup)) + .route("/health/live", get(health_live)) .with_state(Arc::clone(&state)); // Protected routes — gated by Bearer token when a key is configured. @@ -387,24 +389,43 @@ async fn search( })) } -#[derive(Serialize)] -struct HealthResponse { - status: &'static str, - mode: &'static str, - items: u64, +/// Startup probe: always 200 (no migrations to check). +async fn health_startup() -> Json { + Json(serde_json::json!({ "ok": true, "service": "tidaldb" })) } +/// Liveness probe: always 200 (process is alive). +async fn health_live() -> Json { + Json(serde_json::json!({ "ok": true, "service": "tidaldb" })) +} + +/// Readiness probe: 200 when ready, 503 when shutting down. async fn health( State(state): State>, Query(query): Query>, -) -> Result, AppError> { +) -> std::result::Result<(StatusCode, Json), AppError> { + if state.is_shutting_down() { + return Ok(( + StatusCode::SERVICE_UNAVAILABLE, + Json(serde_json::json!({ + "ok": false, + "service": "tidaldb", + "cause": "shutting down" + })), + )); + } + let region = query.get("region").map(|s| s.as_str()); let items = state.item_count(region).map_err(AppError)?; - Ok(Json(HealthResponse { - status: "ok", - mode: "standalone", - items, - })) + + Ok(( + StatusCode::OK, + Json(serde_json::json!({ + "status": "ok", + "mode": "standalone", + "items": items, + })), + )) } struct TidalErrorWrapper(tidaldb::TidalError); diff --git a/tidal-server/src/state.rs b/tidal-server/src/state.rs index b63bc66..5f6c534 100644 --- a/tidal-server/src/state.rs +++ b/tidal-server/src/state.rs @@ -1,5 +1,6 @@ use std::collections::HashMap; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use tidaldb::TidalDb; use tidaldb::query::retrieve::Retrieve; @@ -25,11 +26,23 @@ fn ensure_standalone(region_name: Option<&str>) -> Result<()> { #[derive(Clone)] pub struct ServerState { db: Arc, + shutting_down: Arc, } impl ServerState { pub fn new(db: TidalDb) -> Self { - Self { db: Arc::new(db) } + Self { + db: Arc::new(db), + shutting_down: Arc::new(AtomicBool::new(false)), + } + } + + pub fn set_shutting_down(&self) { + self.shutting_down.store(true, Ordering::SeqCst); + } + + pub fn is_shutting_down(&self) -> bool { + self.shutting_down.load(Ordering::SeqCst) } pub fn write_item( From 82925cee7cb370f2aa5aa638b5b14352e9fcb882 Mon Sep 17 00:00:00 2001 From: Alan Kahn Date: Tue, 3 Mar 2026 10:03:01 -0500 Subject: [PATCH 2/8] feat: add deploy Dockerfile, restore standalone to generic docker/standalone/Dockerfile stays generic (port 9400, baked-in config, auth healthcheck) for anyone running tidalDB independently. docker/deploy/Dockerfile is purpose-built for k8s: port 9500, schema mounted at runtime, persistent data dir, fixed UID for PVC ownership. Co-Authored-By: Claude Opus 4.6 --- docker/deploy/Dockerfile | 31 +++++++++++++++++++++++++++++++ docker/standalone/Dockerfile | 34 +++++++++++++++++++--------------- 2 files changed, 50 insertions(+), 15 deletions(-) create mode 100644 docker/deploy/Dockerfile diff --git a/docker/deploy/Dockerfile b/docker/deploy/Dockerfile new file mode 100644 index 0000000..44ca6fe --- /dev/null +++ b/docker/deploy/Dockerfile @@ -0,0 +1,31 @@ +FROM rust:1.91-slim AS builder +WORKDIR /build +RUN apt-get update && apt-get install -y pkg-config libssl-dev && rm -rf /var/lib/apt/lists/* + +# Copy workspace manifests first for layer caching. +COPY Cargo.toml Cargo.lock ./ +COPY tidal/Cargo.toml tidal/Cargo.toml +COPY tidalctl/Cargo.toml tidalctl/Cargo.toml +COPY tidal-server/Cargo.toml tidal-server/Cargo.toml +COPY applications/forage/engine/Cargo.toml applications/forage/engine/Cargo.toml +COPY applications/forage/server/Cargo.toml applications/forage/server/Cargo.toml +COPY applications/forage/embedder/Cargo.toml applications/forage/embedder/Cargo.toml +COPY applications/iknowyou/engine/Cargo.toml applications/iknowyou/engine/Cargo.toml + +# Copy full workspace and build. +COPY . . +RUN cargo build -p tidal-server --release + +FROM debian:bookworm-slim +RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates curl && rm -rf /var/lib/apt/lists/* +RUN useradd -m -u 10001 tidal +COPY --from=builder --chown=tidal:tidal /build/target/release/tidal-server /usr/local/bin/tidal-server +RUN mkdir -p /config && chown tidal:tidal /config +USER tidal:tidal +WORKDIR /data +EXPOSE 9500 +HEALTHCHECK --interval=30s --timeout=3s --start-period=10s --retries=3 \ + CMD curl -sf http://localhost:9500/health || exit 1 +ENV TIDAL_SERVER_LOG=info +ENTRYPOINT ["/usr/local/bin/tidal-server"] +CMD ["standalone", "--listen", "0.0.0.0:9500", "--schema", "/config/eros-schema.yaml", "--data-dir", "/data"] diff --git a/docker/standalone/Dockerfile b/docker/standalone/Dockerfile index 44ca6fe..8c84fd3 100644 --- a/docker/standalone/Dockerfile +++ b/docker/standalone/Dockerfile @@ -1,6 +1,5 @@ -FROM rust:1.91-slim AS builder -WORKDIR /build -RUN apt-get update && apt-get install -y pkg-config libssl-dev && rm -rf /var/lib/apt/lists/* +FROM rust:1.91 AS builder +WORKDIR /app # Copy workspace manifests first for layer caching. COPY Cargo.toml Cargo.lock ./ @@ -17,15 +16,20 @@ COPY . . RUN cargo build -p tidal-server --release FROM debian:bookworm-slim -RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates curl && rm -rf /var/lib/apt/lists/* -RUN useradd -m -u 10001 tidal -COPY --from=builder --chown=tidal:tidal /build/target/release/tidal-server /usr/local/bin/tidal-server -RUN mkdir -p /config && chown tidal:tidal /config -USER tidal:tidal -WORKDIR /data -EXPOSE 9500 -HEALTHCHECK --interval=30s --timeout=3s --start-period=10s --retries=3 \ - CMD curl -sf http://localhost:9500/health || exit 1 -ENV TIDAL_SERVER_LOG=info -ENTRYPOINT ["/usr/local/bin/tidal-server"] -CMD ["standalone", "--listen", "0.0.0.0:9500", "--schema", "/config/eros-schema.yaml", "--data-dir", "/data"] +WORKDIR /srv +RUN useradd --system --home /srv tidal && \ + apt-get update && apt-get install -y ca-certificates curl && \ + rm -rf /var/lib/apt/lists/* + +COPY --from=builder /app/target/release/tidal-server /usr/local/bin/tidal-server +COPY tidal-server/config /etc/tidal-server + +USER tidal +EXPOSE 9400 9091 + +HEALTHCHECK --interval=30s --timeout=5s --start-period=15s --retries=3 \ + CMD curl -f -H "Authorization: Bearer ${TIDAL_API_KEY:-}" http://localhost:9400/health || exit 1 + +ENTRYPOINT ["tidal-server", "standalone", \ + "--listen", "0.0.0.0:9400", \ + "--metrics", "0.0.0.0:9091"] From c12cde46e8b4775d6736c3475b9dd4a90f5e5739 Mon Sep 17 00:00:00 2001 From: Alan Kahn Date: Tue, 3 Mar 2026 10:06:28 -0500 Subject: [PATCH 3/8] chore: rename schema mount from eros-schema.yaml to schema.yaml Co-Authored-By: Claude Opus 4.6 --- docker/deploy/Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docker/deploy/Dockerfile b/docker/deploy/Dockerfile index 44ca6fe..ef8563f 100644 --- a/docker/deploy/Dockerfile +++ b/docker/deploy/Dockerfile @@ -28,4 +28,4 @@ HEALTHCHECK --interval=30s --timeout=3s --start-period=10s --retries=3 \ CMD curl -sf http://localhost:9500/health || exit 1 ENV TIDAL_SERVER_LOG=info ENTRYPOINT ["/usr/local/bin/tidal-server"] -CMD ["standalone", "--listen", "0.0.0.0:9500", "--schema", "/config/eros-schema.yaml", "--data-dir", "/data"] +CMD ["standalone", "--listen", "0.0.0.0:9500", "--schema", "/config/schema.yaml", "--data-dir", "/data"] From 16214ebfcbeb34d24d634ef731bbad11532d51dd Mon Sep 17 00:00:00 2001 From: Alan Kahn Date: Tue, 3 Mar 2026 10:26:40 -0500 Subject: [PATCH 4/8] chore: add ok/service fields to readiness response for consistency Matches the response shape used by all other services in the infrastructure (ok, service, cause on failure). Co-Authored-By: Claude Opus 4.6 --- tidal-server/src/router.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tidal-server/src/router.rs b/tidal-server/src/router.rs index fee1eb8..62a5cde 100644 --- a/tidal-server/src/router.rs +++ b/tidal-server/src/router.rs @@ -421,7 +421,8 @@ async fn health( Ok(( StatusCode::OK, Json(serde_json::json!({ - "status": "ok", + "ok": true, + "service": "tidaldb", "mode": "standalone", "items": items, })), From dd6c709cbedd9567b8c10dac26528fa68aa3d1af Mon Sep 17 00:00:00 2001 From: Alan Kahn Date: Tue, 3 Mar 2026 12:25:58 -0500 Subject: [PATCH 5/8] feat: support PORT env var for listen address Clap now reads PORT from the environment, accepting either a bare port number (e.g. 8080 -> 0.0.0.0:8080) or a full host:port. CLI --listen flag still takes precedence. Deploy Dockerfile defaults PORT=9500 and removes the hardcoded --listen argument. Co-Authored-By: Claude Opus 4.6 --- docker/deploy/Dockerfile | 5 +++-- tidal-server/Cargo.toml | 2 +- tidal-server/src/main.rs | 16 +++++++++++++++- 3 files changed, 19 insertions(+), 4 deletions(-) diff --git a/docker/deploy/Dockerfile b/docker/deploy/Dockerfile index ef8563f..87e341c 100644 --- a/docker/deploy/Dockerfile +++ b/docker/deploy/Dockerfile @@ -25,7 +25,8 @@ USER tidal:tidal WORKDIR /data EXPOSE 9500 HEALTHCHECK --interval=30s --timeout=3s --start-period=10s --retries=3 \ - CMD curl -sf http://localhost:9500/health || exit 1 + CMD curl -sf http://localhost:${PORT:-9500}/health || exit 1 ENV TIDAL_SERVER_LOG=info +ENV PORT=9500 ENTRYPOINT ["/usr/local/bin/tidal-server"] -CMD ["standalone", "--listen", "0.0.0.0:9500", "--schema", "/config/schema.yaml", "--data-dir", "/data"] +CMD ["standalone", "--schema", "/config/schema.yaml", "--data-dir", "/data"] diff --git a/tidal-server/Cargo.toml b/tidal-server/Cargo.toml index a6171c7..fc1ad76 100644 --- a/tidal-server/Cargo.toml +++ b/tidal-server/Cargo.toml @@ -11,7 +11,7 @@ path = "src/lib.rs" [dependencies] axum = "0.8" -clap = { version = "4.5", features = ["derive"] } +clap = { version = "4.5", features = ["derive", "env"] } subtle = "2" tower = { version = "0.5", features = ["limit"] } tower-http = { version = "0.6", features = ["timeout", "trace", "request-id"] } diff --git a/tidal-server/src/main.rs b/tidal-server/src/main.rs index f50be46..f06d8c3 100644 --- a/tidal-server/src/main.rs +++ b/tidal-server/src/main.rs @@ -24,7 +24,7 @@ enum Command { #[derive(Args)] struct StandaloneArgs { - #[arg(long, default_value = "127.0.0.1:9400")] + #[arg(long, default_value = "127.0.0.1:9400", env = "PORT", value_parser = parse_listen_addr)] listen: String, #[arg(long)] schema: Option, @@ -54,6 +54,20 @@ async fn run() -> Result<()> { } } +/// Parse a listen address from either a full `host:port` string or a bare port number. +/// When `PORT=8080` is set, clap passes `"8080"` — this normalises it to `0.0.0.0:8080`. +fn parse_listen_addr(s: &str) -> std::result::Result { + if s.contains(':') { + s.parse::() + .map(|_| s.to_string()) + .map_err(|e| format!("invalid address '{s}': {e}")) + } else { + s.parse::() + .map(|port| format!("0.0.0.0:{port}")) + .map_err(|_| format!("expected host:port or port number, got '{s}'")) + } +} + fn init_tracing() { let env_filter = std::env::var("TIDAL_SERVER_LOG").unwrap_or_else(|_| "info".into()); let _ = tracing_subscriber::fmt() From 889c746bc15bd0651b8cd2ef9365f624774a91d4 Mon Sep 17 00:00:00 2001 From: Alan Kahn Date: Tue, 3 Mar 2026 14:53:48 -0500 Subject: [PATCH 6/8] fix: add g++ to deploy Dockerfile for cc-rs compilation rust:1.91-slim doesn't include a C++ compiler, which cc-rs needs. The standalone/cluster Dockerfiles use the full rust:1.91 image so they weren't affected. Co-Authored-By: Claude Opus 4.6 --- docker/deploy/Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docker/deploy/Dockerfile b/docker/deploy/Dockerfile index 87e341c..03108ca 100644 --- a/docker/deploy/Dockerfile +++ b/docker/deploy/Dockerfile @@ -1,6 +1,6 @@ FROM rust:1.91-slim AS builder WORKDIR /build -RUN apt-get update && apt-get install -y pkg-config libssl-dev && rm -rf /var/lib/apt/lists/* +RUN apt-get update && apt-get install -y pkg-config libssl-dev g++ && rm -rf /var/lib/apt/lists/* # Copy workspace manifests first for layer caching. COPY Cargo.toml Cargo.lock ./ From 81bb7b24e77a8280152e0aa1546d2057efb2619d Mon Sep 17 00:00:00 2001 From: Alan Kahn Date: Tue, 3 Mar 2026 16:41:28 -0500 Subject: [PATCH 7/8] fix: rebuild item indexes from durable storage on restart After a pod restart, in-memory indexes (universe bitmap, category, format, creator, tags, duration, created_at) were empty because they were only populated by write_item_with_metadata() calls. This caused /health to report items: 0 and queries to return no results despite data persisting on disk in fjall storage. Add rebuild_item_indexes() which scans the items keyspace for Tag::Meta entries on startup and repopulates all indexes from stored metadata. Update m2_uat crash recovery test to assert indexes survive restart without rewriting items. Co-Authored-By: Claude Opus 4.6 --- tidal/src/db/open.rs | 16 +++++- tidal/src/db/state_rebuild.rs | 92 +++++++++++++++++++++++++++++++++++ tidal/tests/m2_uat.rs | 32 +++++++++--- 3 files changed, 132 insertions(+), 8 deletions(-) diff --git a/tidal/src/db/open.rs b/tidal/src/db/open.rs index 3d5f015..e9dab8f 100644 --- a/tidal/src/db/open.rs +++ b/tidal/src/db/open.rs @@ -22,7 +22,7 @@ use super::config::StorageMode; use super::storage_box::StorageBox; use super::wal_bridge::WalHandleWriter; -use super::state_rebuild::rebuild_entity_state; +use super::state_rebuild::{rebuild_entity_state, rebuild_item_indexes}; /// Bundle returned by [`TidalDb::open_with_schema`] to avoid a fragile tuple. /// @@ -177,6 +177,20 @@ impl super::TidalDb { let interaction_ledger = InteractionLedger::new(); rebuild_entity_state(&storage, &user_state, &creator_items, &interaction_ledger)?; + // Rebuild item indexes (universe bitmap + bitmap/range indexes) + // from persisted metadata so queries and /health work immediately. + let mut universe = universe; + rebuild_item_indexes( + &storage, + &mut universe, + &category_index, + &format_index, + &creator_index, + &tag_index, + &duration_index, + &created_at_index, + )?; + // Read preference vector dimensionality from the schema. let pref_dim = schema_def .embedding_slots() diff --git a/tidal/src/db/state_rebuild.rs b/tidal/src/db/state_rebuild.rs index b512cf5..e630d27 100644 --- a/tidal/src/db/state_rebuild.rs +++ b/tidal/src/db/state_rebuild.rs @@ -5,10 +5,14 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::Duration; +use roaring::RoaringBitmap; + use crate::cohort::CohortSignalLedger; use crate::query::suggest::SuggestionIndex; use crate::schema::{TidalError, Timestamp}; use crate::signals::{DEFAULT_MAX_SIGNAL_ENTRIES, SignalLedger, trim_cold_entries}; +use crate::storage::indexes::bitmap::BitmapIndex; +use crate::storage::indexes::range::RangeIndex; use crate::storage::{StorageEngine, Tag}; use super::metadata::deserialize_metadata; @@ -185,6 +189,94 @@ pub(super) fn rebuild_suggestion_index(storage: &StorageBox, suggestion_index: & } } +/// Rebuild item indexes (universe bitmap + bitmap/range indexes) from durable storage. +/// +/// Scans the items keyspace for `Tag::Meta` keys and populates the in-memory +/// universe bitmap and all six item indexes (category, format, creator, tags, +/// duration, `created_at`) from the persisted metadata. This ensures `/health` +/// reports the correct item count and queries work immediately after a restart +/// without rewriting items. +/// +/// Items without explicit `created_at` metadata will not have a `created_at` +/// range index entry restored — the original `Timestamp::now()` default was +/// only stored in the range index, not in metadata. +/// +/// For ephemeral mode the engine is empty, so this is a no-op. +#[allow(clippy::cast_possible_truncation, clippy::too_many_arguments)] +pub(super) fn rebuild_item_indexes( + storage: &StorageBox, + universe: &mut RoaringBitmap, + category_index: &BitmapIndex, + format_index: &BitmapIndex, + creator_index: &BitmapIndex, + tag_index: &BitmapIndex, + duration_index: &RangeIndex, + created_at_index: &RangeIndex, +) -> crate::Result<()> { + use crate::storage::keys::parse_key; + + let mut count = 0u64; + let scan_start = std::time::Instant::now(); + + for entry in storage.items_engine().scan_prefix(&[]) { + let (key, value) = entry.map_err(TidalError::from)?; + + if let Some((entity_id, Tag::Meta, _suffix)) = parse_key(&key) { + let id_u32 = entity_id.as_u64() as u32; + let meta = deserialize_metadata(&value); + + // Universe bitmap. + universe.insert(id_u32); + + // Bitmap indexes. + if let Some(val) = meta.get("category") { + category_index.insert(id_u32, val); + } + if let Some(val) = meta.get("format") { + format_index.insert(id_u32, val); + } + if let Some(val) = meta.get("creator_id") { + creator_index.insert(id_u32, val); + } + if let Some(tags) = meta.get("tags") { + for tag in tags.split(',') { + let tag = tag.trim(); + if !tag.is_empty() { + tag_index.insert(id_u32, tag); + } + } + } + + // Range indexes. + if let Some(val) = meta.get("duration") + && let Ok(dur) = val.parse::() + { + duration_index.insert(id_u32, dur); + } + if let Some(val) = meta.get("created_at") + && let Ok(ts) = val.parse::() + { + created_at_index.insert(id_u32, ts); + } + + count += 1; + if count.is_multiple_of(10_000) { + tracing::info!(rebuilt = count, "item index rebuild in progress"); + } + } + } + + if count > 0 { + tracing::info!( + items = count, + elapsed_ms = scan_start.elapsed().as_millis(), + "item indexes rebuilt from durable storage" + ); + } + + Ok(()) +} + /// Background thread body: checkpoint signal state to storage every 30 seconds. /// /// Checkpoints both the global signal ledger and the cohort signal ledger diff --git a/tidal/tests/m2_uat.rs b/tidal/tests/m2_uat.rs index 4046c28..60b5faf 100644 --- a/tidal/tests/m2_uat.rs +++ b/tidal/tests/m2_uat.rs @@ -415,9 +415,14 @@ fn milestone_2_uat() { .open() .expect("reopen failed (second session)"); - // Note: In-memory indexes (bitmaps, range) are NOT persisted in M2. - // After reopen, the universe is empty and queries return no results. - // Signal state IS persisted via WAL + checkpoint. + // Item indexes are now rebuilt from durable storage on restart. + // Universe bitmap, bitmap indexes, and range indexes should all + // be populated without rewriting items. + assert_eq!( + db2.item_count(), + 1_000, + "crash recovery: universe must contain 1K items after reopen (rebuilt from storage)" + ); // Verify signal recovery: decay score should survive. let score_after_reopen = db2 @@ -444,11 +449,24 @@ fn milestone_2_uat() { "crash recovery: share count for entity 42 should be >= 100, got {share_count_recovered}" ); - // Rewrite items so the in-memory indexes are repopulated, then re-query. - write_items(&db2, base_ns); - assert_eq!(db2.item_count(), 1_000, "universe must be repopulated"); + // Verify filtered query works without rewriting items — bitmap indexes + // were rebuilt from storage. + let query = Retrieve::builder() + .profile("hot") + .limit(10) + .filter(FilterExpr::CategoryEq("jazz".into())) + .build() + .expect("post-recovery filtered query build failed"); + let results = db2 + .retrieve(&query) + .expect("post-recovery filtered query failed"); + assert!( + !results.items.is_empty(), + "post-recovery filtered query must return results (indexes rebuilt from storage)" + ); + assert_score_descending(&results.items, "post-recovery hot+jazz"); - // Verify query still works after recovery + index repopulation. + // Verify unfiltered query also works. let query = Retrieve::builder() .profile("new") .limit(10) From 1d826c87b2636a49f55672386f4207c8218500a6 Mon Sep 17 00:00:00 2001 From: Alan Kahn Date: Wed, 4 Mar 2026 12:12:54 -0500 Subject: [PATCH 8/8] feat: schema-level ranking profile definitions Add support for defining ranking profiles in schema YAML, allowing deployments to override builtin profiles with deployment-specific signal names and tuning parameters. - Add override_register() to ProfileRegistry for clean builtin replacement - Add with_profiles() to TidalDbBuilder to thread schema profiles - Parse profiles section in config.rs with full sort/strategy/agg support - Validate that profile signal references exist in schema at startup - Change load_schema() to return (Schema, Vec) This closes the gap where the builtin for_you profile referenced signals (view, like, share) that don't exist in deployment schemas, causing all feed scores to normalize to 1.0. Co-Authored-By: Claude Opus 4.6 --- Cargo.lock | 1 + tidal-server/Cargo.toml | 3 + tidal-server/src/config.rs | 631 ++++++++++++++++++++++++++++++- tidal-server/src/main.rs | 6 +- tidal-server/tests/middleware.rs | 3 +- tidal/src/db/builder.rs | 18 +- tidal/src/db/builder_tests.rs | 1 + tidal/src/db/open.rs | 13 + tidal/src/ranking/registry.rs | 104 +++++ 9 files changed, 770 insertions(+), 10 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1bb7cc1..6cc3849 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3325,6 +3325,7 @@ dependencies = [ "serde_json", "serde_yaml", "subtle", + "tempfile", "thiserror 2.0.18", "tidaldb", "tokio", diff --git a/tidal-server/Cargo.toml b/tidal-server/Cargo.toml index fc1ad76..7281c4e 100644 --- a/tidal-server/Cargo.toml +++ b/tidal-server/Cargo.toml @@ -23,3 +23,6 @@ tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } tidaldb = { path = "../tidal" } + +[dev-dependencies] +tempfile = "3" diff --git a/tidal-server/src/config.rs b/tidal-server/src/config.rs index 933b821..311ab79 100644 --- a/tidal-server/src/config.rs +++ b/tidal-server/src/config.rs @@ -3,13 +3,17 @@ use std::path::Path; use std::time::Duration; use serde::Deserialize; +use tidaldb::ranking::profile::{ + Boost, CandidateStrategy, DiversitySpec, Exclude, Gate, Penalty, ProfileDecay, RankingProfile, + SignalAgg, Sort, +}; use tidaldb::schema::{DecaySpec, EntityKind, Schema, SchemaBuilder, TextFieldType, Window}; use crate::error::{Result, ServerError}; const DEFAULT_SCHEMA_YAML: &str = include_str!("../config/default-schema.yaml"); -pub fn load_schema(path: Option<&Path>) -> Result { +pub fn load_schema(path: Option<&Path>) -> Result<(Schema, Vec)> { let raw = read_config(path, DEFAULT_SCHEMA_YAML)?; let spec: SchemaSpec = serde_yaml::from_str(&raw) .map_err(|e| ServerError::SchemaConfig(format!("parse schema yaml: {e}")))?; @@ -20,14 +24,17 @@ pub fn load_schema(path: Option<&Path>) -> Result { )); } + // Collect signal names for profile validation. + let signal_names: Vec = spec.signals.iter().map(|s| s.name.clone()).collect(); + let mut builder = SchemaBuilder::new(); - for signal in spec.signals { + for signal in &spec.signals { let mut sig = builder.signal( &signal.name, parse_entity_kind(&signal.entity)?, signal.decay.to_decay_spec()?, ); - if let Some(windows) = signal.windows { + if let Some(ref windows) = signal.windows { let parsed: Result> = windows.iter().map(|w| parse_window(w)).collect(); sig = sig.windows(&parsed?); } @@ -37,13 +44,13 @@ pub fn load_schema(path: Option<&Path>) -> Result { let _ = sig.add(); } - if let Some(text_fields) = spec.text_fields { + if let Some(ref text_fields) = spec.text_fields { for field in text_fields { builder.text_field(&field.name, parse_text_field_type(&field.kind)?); } } - if let Some(embeddings) = spec.embedding_slots { + if let Some(ref embeddings) = spec.embedding_slots { for slot in embeddings { builder.embedding_slot( &slot.name, @@ -53,7 +60,22 @@ pub fn load_schema(path: Option<&Path>) -> Result { } } - builder.build().map_err(ServerError::SchemaBuild) + let schema = builder.build().map_err(ServerError::SchemaBuild)?; + + // Parse profiles and validate signal references. + let profiles = if let Some(profile_specs) = spec.profiles { + let mut parsed = Vec::with_capacity(profile_specs.len()); + for spec in profile_specs { + let profile = spec.to_ranking_profile()?; + validate_profile_signals(&profile, &signal_names)?; + parsed.push(profile); + } + parsed + } else { + Vec::new() + }; + + Ok((schema, profiles)) } fn read_config(path: Option<&Path>, fallback: &str) -> Result { @@ -70,6 +92,8 @@ struct SchemaSpec { text_fields: Option>, #[serde(default)] embedding_slots: Option>, + #[serde(default)] + profiles: Option>, } #[derive(Debug, Deserialize)] @@ -171,3 +195,598 @@ fn parse_text_field_type(input: &str) -> Result { ))), } } + +// ── Profile YAML types ────────────────────────────────────────────────────── + +#[derive(Debug, Deserialize)] +struct ProfileSpec { + name: String, + #[serde(default = "default_version")] + version: u32, + #[serde(default)] + candidate_strategy: Option, + #[serde(default)] + sort: Option, + #[serde(default)] + boosts: Vec, + #[serde(default)] + decay: Option, + #[serde(default)] + gates: Vec, + #[serde(default)] + penalties: Vec, + #[serde(default)] + excludes: Vec, + #[serde(default)] + diversity: Option, + #[serde(default)] + exploration: f64, +} + +fn default_version() -> u32 { + 1 +} + +#[derive(Debug, Deserialize)] +struct BoostSpec { + signal: String, + agg: String, + window: String, + weight: f64, +} + +#[derive(Debug, Deserialize)] +struct GateSpec { + signal: String, + agg: String, + window: String, + min_threshold: f64, +} + +#[derive(Debug, Deserialize)] +struct PenaltySpec { + signal: String, + agg: String, + window: String, + weight: f64, +} + +#[derive(Debug, Deserialize)] +struct ExcludeSpec { + signal: String, + agg: String, + window: String, + above: f64, +} + +#[derive(Debug, Deserialize)] +struct ProfileDecaySpec { + signal: String, + half_life_secs: u64, + weight: f64, +} + +#[derive(Debug, Deserialize)] +struct DiversitySpecConfig { + #[serde(default)] + max_per_creator: Option, + #[serde(default)] + format_mix_max_fraction: Option, +} + +/// Sort can be a simple string (`"trending"`, `"new"`) or a map with parameters. +#[derive(Debug, Deserialize)] +#[serde(untagged)] +enum SortSpec { + Simple(String), + Parameterized(std::collections::HashMap), +} + +/// Candidate strategy can be a simple string or a map with parameters. +#[derive(Debug, Deserialize)] +#[serde(untagged)] +enum CandidateStrategySpec { + Simple(String), + Parameterized(std::collections::HashMap), +} + +// ── Profile conversion ────────────────────────────────────────────────────── + +impl ProfileSpec { + fn to_ranking_profile(&self) -> Result { + let candidate_strategy = match &self.candidate_strategy { + Some(spec) => parse_candidate_strategy(spec)?, + None => CandidateStrategy::Scan { + sort_field: "created_at".into(), + }, + }; + + let sort = match &self.sort { + Some(spec) => Some(parse_sort(spec)?), + None => None, + }; + + let boosts = self + .boosts + .iter() + .map(|b| { + Ok(Boost { + signal: b.signal.clone(), + agg: parse_agg(&b.agg)?, + window: parse_window(&b.window)?, + weight: b.weight, + }) + }) + .collect::>>()?; + + let gates = self + .gates + .iter() + .map(|g| { + Ok(Gate { + signal: g.signal.clone(), + agg: parse_agg(&g.agg)?, + window: parse_window(&g.window)?, + min_threshold: g.min_threshold, + }) + }) + .collect::>>()?; + + let penalties = self + .penalties + .iter() + .map(|p| { + Ok(Penalty { + signal: p.signal.clone(), + agg: parse_agg(&p.agg)?, + window: parse_window(&p.window)?, + weight: p.weight, + }) + }) + .collect::>>()?; + + let excludes = self + .excludes + .iter() + .map(|e| { + Ok(Exclude { + signal: e.signal.clone(), + agg: parse_agg(&e.agg)?, + window: parse_window(&e.window)?, + above: e.above, + }) + }) + .collect::>>()?; + + let decay = self.decay.as_ref().map(|d| ProfileDecay { + signal: d.signal.clone(), + half_life_secs: d.half_life_secs, + weight: d.weight, + }); + + let diversity = match &self.diversity { + Some(d) => DiversitySpec { + max_per_creator: d.max_per_creator, + format_mix_max_fraction: d.format_mix_max_fraction, + }, + None => DiversitySpec::default(), + }; + + Ok(RankingProfile { + name: self.name.clone(), + version: self.version, + candidate_strategy, + boosts, + decay, + gates, + penalties, + excludes, + diversity, + exploration: self.exploration, + sort, + is_builtin: false, + }) + } +} + +fn parse_agg(input: &str) -> Result { + match input.trim().to_lowercase().as_str() { + "value" => Ok(SignalAgg::Value), + "velocity" => Ok(SignalAgg::Velocity), + "decay_score" => Ok(SignalAgg::DecayScore), + "ratio" => Ok(SignalAgg::Ratio), + "relative_velocity" => Ok(SignalAgg::RelativeVelocity), + other => Err(ServerError::SchemaConfig(format!( + "unknown signal aggregation '{other}'" + ))), + } +} + +fn parse_sort(spec: &SortSpec) -> Result { + match spec { + SortSpec::Simple(s) => match s.trim().to_lowercase().as_str() { + "trending" => Ok(Sort::Trending), + "rising" => Ok(Sort::Rising), + "controversial" => Ok(Sort::Controversial), + "hidden_gems" => Ok(Sort::HiddenGems), + "shuffle" => Ok(Sort::Shuffle), + "new" => Ok(Sort::New), + "most_followed" => Ok(Sort::MostFollowed), + "creator_engagement_rate" => Ok(Sort::CreatorEngagementRate), + "alphabetical_asc" => Ok(Sort::AlphabeticalAsc), + "alphabetical_desc" => Ok(Sort::AlphabeticalDesc), + "shortest" => Ok(Sort::Shortest), + "longest" => Ok(Sort::Longest), + "live_viewer_count" => Ok(Sort::LiveViewerCount), + "date_saved" => Ok(Sort::DateSaved), + other => Err(ServerError::SchemaConfig(format!( + "unknown sort mode '{other}'" + ))), + }, + SortSpec::Parameterized(map) => { + if let Some(val) = map.get("hot") { + let gravity = extract_f64(val, "gravity")?.unwrap_or(1.8); + Ok(Sort::Hot { gravity }) + } else if let Some(val) = map.get("top_window") { + let window = extract_window(val, "window")?; + Ok(Sort::TopWindow { window }) + } else if let Some(val) = map.get("most_viewed") { + let window = extract_window(val, "window")?; + Ok(Sort::MostViewed { window }) + } else if let Some(val) = map.get("most_liked") { + let window = extract_window(val, "window")?; + Ok(Sort::MostLiked { window }) + } else if let Some(val) = map.get("most_commented") { + let window = extract_window(val, "window")?; + Ok(Sort::MostCommented { window }) + } else if let Some(val) = map.get("most_shared") { + let window = extract_window(val, "window")?; + Ok(Sort::MostShared { window }) + } else { + Err(ServerError::SchemaConfig(format!( + "unknown parameterized sort: {map:?}" + ))) + } + } + } +} + +fn parse_candidate_strategy(spec: &CandidateStrategySpec) -> Result { + match spec { + CandidateStrategySpec::Simple(s) => match s.trim().to_lowercase().as_str() { + "scan" => Ok(CandidateStrategy::Scan { + sort_field: "created_at".into(), + }), + "relationship" => Ok(CandidateStrategy::Relationship), + "hybrid" => Ok(CandidateStrategy::Hybrid), + "cohort_trending" => Ok(CandidateStrategy::CohortTrending), + other => Err(ServerError::SchemaConfig(format!( + "unknown candidate strategy '{other}'" + ))), + }, + CandidateStrategySpec::Parameterized(map) => { + if let Some(val) = map.get("ann") { + let slot = extract_string(val, "slot")?; + let limit = extract_u64(val, "limit")?.unwrap_or(100) as usize; + Ok(CandidateStrategy::Ann { slot, limit }) + } else if let Some(val) = map.get("signal_ranked") { + let signal = extract_string(val, "signal")?; + let window_str = extract_string_field(val, "window")?; + let window = parse_window(&window_str)?; + Ok(CandidateStrategy::SignalRanked { signal, window }) + } else if let Some(val) = map.get("scan") { + let sort_field = + extract_string_field(val, "sort_field").unwrap_or_else(|_| "created_at".into()); + Ok(CandidateStrategy::Scan { sort_field }) + } else { + Err(ServerError::SchemaConfig(format!( + "unknown parameterized candidate strategy: {map:?}" + ))) + } + } + } +} + +// ── YAML value extraction helpers ─────────────────────────────────────────── + +fn extract_f64(val: &serde_yaml::Value, field: &str) -> Result> { + match val { + serde_yaml::Value::Mapping(map) => { + if let Some(v) = map.get(serde_yaml::Value::String(field.to_owned())) { + v.as_f64() + .or_else(|| v.as_i64().map(|i| i as f64)) + .map(Some) + .ok_or_else(|| ServerError::SchemaConfig(format!("'{field}' must be a number"))) + } else { + Ok(None) + } + } + _ => Ok(None), + } +} + +fn extract_u64(val: &serde_yaml::Value, field: &str) -> Result> { + match val { + serde_yaml::Value::Mapping(map) => { + if let Some(v) = map.get(serde_yaml::Value::String(field.to_owned())) { + v.as_u64().map(Some).ok_or_else(|| { + ServerError::SchemaConfig(format!("'{field}' must be a positive integer")) + }) + } else { + Ok(None) + } + } + _ => Ok(None), + } +} + +fn extract_string(val: &serde_yaml::Value, field: &str) -> Result { + extract_string_field(val, field) +} + +fn extract_string_field(val: &serde_yaml::Value, field: &str) -> Result { + match val { + serde_yaml::Value::Mapping(map) => { + if let Some(v) = map.get(serde_yaml::Value::String(field.to_owned())) { + v.as_str() + .map(|s| s.to_owned()) + .ok_or_else(|| ServerError::SchemaConfig(format!("'{field}' must be a string"))) + } else { + Err(ServerError::SchemaConfig(format!( + "missing required field '{field}'" + ))) + } + } + _ => Err(ServerError::SchemaConfig(format!( + "expected mapping with field '{field}'" + ))), + } +} + +fn extract_window(val: &serde_yaml::Value, field: &str) -> Result { + let s = extract_string_field(val, field)?; + parse_window(&s) +} + +// ── Signal validation ─────────────────────────────────────────────────────── + +fn validate_profile_signals(profile: &RankingProfile, signal_names: &[String]) -> Result<()> { + let check = |signal: &str, context: &str| -> Result<()> { + if !signal_names.iter().any(|s| s == signal) { + Err(ServerError::SchemaConfig(format!( + "profile '{}': {} references unknown signal '{signal}'", + profile.name, context + ))) + } else { + Ok(()) + } + }; + + for boost in &profile.boosts { + check(&boost.signal, "boost")?; + } + for gate in &profile.gates { + check(&gate.signal, "gate")?; + } + for penalty in &profile.penalties { + check(&penalty.signal, "penalty")?; + } + for exclude in &profile.excludes { + check(&exclude.signal, "exclude")?; + } + if let Some(ref decay) = profile.decay { + check(&decay.signal, "decay")?; + } + Ok(()) +} + +// ── Tests ─────────────────────────────────────────────────────────────────── + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use super::*; + + #[test] + fn parse_schema_with_profiles() { + let yaml = r#" +signals: + - name: chat + entity: item + decay: + exponential: + half_life_seconds: 1209600 + windows: [24h, 7d] + - name: favorite + entity: item + decay: + exponential: + half_life_seconds: 2592000 + windows: [7d, 30d] + +profiles: + - name: for_you + sort: + hot: + gravity: 1.5 + boosts: + - signal: chat + agg: decay_score + window: all_time + weight: 1.5 + - signal: favorite + agg: decay_score + window: all_time + weight: 2.5 + diversity: + max_per_creator: 2 + exploration: 0.1 +"#; + let spec: SchemaSpec = serde_yaml::from_str(yaml).unwrap(); + assert!(spec.profiles.is_some()); + let profiles = spec.profiles.unwrap(); + assert_eq!(profiles.len(), 1); + let profile = profiles[0].to_ranking_profile().unwrap(); + assert_eq!(profile.name, "for_you"); + assert_eq!(profile.boosts.len(), 2); + assert!(!profile.is_builtin); + assert!((profile.exploration - 0.1).abs() < f64::EPSILON); + assert!( + matches!(profile.sort, Some(Sort::Hot { gravity }) if (gravity - 1.5).abs() < f64::EPSILON) + ); + assert_eq!(profile.diversity.max_per_creator, Some(2)); + } + + #[test] + fn parse_simple_sort_modes() { + let cases = vec![ + ("trending", true), + ("new", true), + ("shuffle", true), + ("rising", true), + ("controversial", true), + ("hidden_gems", true), + ("bad_sort", false), + ]; + for (input, should_ok) in cases { + let result = parse_sort(&SortSpec::Simple(input.into())); + assert_eq!(result.is_ok(), should_ok, "sort '{input}'"); + } + } + + #[test] + fn parse_parameterized_sort() { + let mut map = std::collections::HashMap::new(); + let mut inner = serde_yaml::Mapping::new(); + inner.insert( + serde_yaml::Value::String("window".into()), + serde_yaml::Value::String("7d".into()), + ); + map.insert("top_window".into(), serde_yaml::Value::Mapping(inner)); + let result = parse_sort(&SortSpec::Parameterized(map)).unwrap(); + assert!(matches!( + result, + Sort::TopWindow { + window: Window::SevenDays + } + )); + } + + #[test] + fn parse_agg_variants() { + assert!(matches!(parse_agg("value").unwrap(), SignalAgg::Value)); + assert!(matches!( + parse_agg("velocity").unwrap(), + SignalAgg::Velocity + )); + assert!(matches!( + parse_agg("decay_score").unwrap(), + SignalAgg::DecayScore + )); + assert!(matches!(parse_agg("ratio").unwrap(), SignalAgg::Ratio)); + assert!(matches!( + parse_agg("relative_velocity").unwrap(), + SignalAgg::RelativeVelocity + )); + assert!(parse_agg("unknown").is_err()); + } + + #[test] + fn parse_candidate_strategy_variants() { + let scan = parse_candidate_strategy(&CandidateStrategySpec::Simple("scan".into())).unwrap(); + assert!(matches!(scan, CandidateStrategy::Scan { .. })); + + let rel = parse_candidate_strategy(&CandidateStrategySpec::Simple("relationship".into())) + .unwrap(); + assert!(matches!(rel, CandidateStrategy::Relationship)); + } + + #[test] + fn unknown_signal_in_profile_rejected() { + let profile = RankingProfile { + name: "test".into(), + version: 1, + candidate_strategy: CandidateStrategy::Scan { + sort_field: "created_at".into(), + }, + boosts: vec![Boost { + signal: "nonexistent".into(), + agg: SignalAgg::Value, + window: Window::AllTime, + weight: 1.0, + }], + decay: None, + gates: vec![], + penalties: vec![], + excludes: vec![], + diversity: DiversitySpec::default(), + exploration: 0.0, + sort: None, + is_builtin: false, + }; + let signal_names = vec!["chat".into(), "favorite".into()]; + let result = validate_profile_signals(&profile, &signal_names); + assert!(result.is_err()); + let err = result.unwrap_err().to_string(); + assert!(err.contains("nonexistent"), "error: {err}"); + } + + #[test] + fn load_schema_returns_tuple() { + let (schema, profiles) = load_schema(None).unwrap(); + // Default schema has no profiles section. + assert!(profiles.is_empty()); + assert!(schema.signals().next().is_some()); + } + + #[test] + fn schema_with_no_profiles_section() { + let yaml = r#" +signals: + - name: view + entity: item + decay: + exponential: + half_life_seconds: 604800 + windows: [24h, 7d] +"#; + let tmp = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(tmp.path(), yaml).unwrap(); + let (_, profiles) = load_schema(Some(tmp.path())).unwrap(); + assert!(profiles.is_empty()); + } + + #[test] + fn profile_defaults() { + let yaml = r#" +signals: + - name: view + entity: item + decay: + exponential: + half_life_seconds: 604800 + windows: [24h, 7d] + +profiles: + - name: minimal + boosts: + - signal: view + agg: value + window: all_time + weight: 1.0 +"#; + let tmp = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(tmp.path(), yaml).unwrap(); + let (_, profiles) = load_schema(Some(tmp.path())).unwrap(); + assert_eq!(profiles.len(), 1); + let p = &profiles[0]; + assert_eq!(p.version, 1); + assert!(!p.is_builtin); + assert!(p.sort.is_none()); + assert!((p.exploration).abs() < f64::EPSILON); + assert!(matches!( + p.candidate_strategy, + CandidateStrategy::Scan { .. } + )); + } +} diff --git a/tidal-server/src/main.rs b/tidal-server/src/main.rs index f06d8c3..a236674 100644 --- a/tidal-server/src/main.rs +++ b/tidal-server/src/main.rs @@ -76,9 +76,11 @@ fn init_tracing() { } async fn run_standalone(args: StandaloneArgs) -> Result<()> { - let schema = load_schema(args.schema.as_deref())?; + let (schema, profiles) = load_schema(args.schema.as_deref())?; - let mut builder = TidalDb::builder().with_schema(schema.clone()); + let mut builder = TidalDb::builder() + .with_schema(schema.clone()) + .with_profiles(profiles); if let Some(dir) = args.data_dir { builder = builder.with_data_dir(dir); } else { diff --git a/tidal-server/tests/middleware.rs b/tidal-server/tests/middleware.rs index f198917..e1a7a21 100644 --- a/tidal-server/tests/middleware.rs +++ b/tidal-server/tests/middleware.rs @@ -17,10 +17,11 @@ use tower::ServiceExt; fn make_state() -> Arc { // Use the default schema so ranking profiles are available — same setup as // `run_standalone` in main.rs. - let schema = tidal_server::config::load_schema(None).unwrap(); + let (schema, profiles) = tidal_server::config::load_schema(None).unwrap(); let db = TidalDb::builder() .ephemeral() .with_schema(schema) + .with_profiles(profiles) .open() .unwrap(); Arc::new(ServerState::new(db)) diff --git a/tidal/src/db/builder.rs b/tidal/src/db/builder.rs index dfa1732..42f31ce 100644 --- a/tidal/src/db/builder.rs +++ b/tidal/src/db/builder.rs @@ -10,6 +10,7 @@ use super::metrics::MetricsState; use super::paths::Paths; use crate::load::RateLimiterConfig; +use crate::ranking::profile::RankingProfile; use crate::schema::Schema; /// Fluent builder for constructing a [`TidalDb`] instance. @@ -55,6 +56,8 @@ pub struct TidalDbBuilder { /// When set and `NodeRole::Follower`, the receiver is started automatically. /// When set and `NodeRole::Leader`, the WAL shipper is spawned automatically. transport: Option>, + /// Schema-defined profiles that override matching builtins. + profiles: Vec, } impl TidalDbBuilder { @@ -67,6 +70,7 @@ impl TidalDbBuilder { schema: None, rate_limiter_config: None, transport: None, + profiles: Vec::new(), } } @@ -84,6 +88,18 @@ impl TidalDbBuilder { self } + /// Attach schema-defined ranking profiles that override matching builtins. + /// + /// Profiles provided here are registered via [`override_register`] after + /// built-in profiles during [`open`](Self::open). This allows schema YAML + /// to replace built-in profile definitions with deployment-specific signal + /// names and tuning parameters. + #[must_use] + pub fn with_profiles(mut self, profiles: Vec) -> Self { + self.profiles = profiles; + self + } + /// Configure per-agent session rate limiting. /// /// When set, each `(agent_id, session_id)` pair gets a token bucket with @@ -328,7 +344,7 @@ impl TidalDbBuilder { if let Some(schema) = self.schema { // Wire in storage, WAL, signal ledger, and M2 indexes. - let result = TidalDb::open_with_schema(&self.config, schema)?; + let result = TidalDb::open_with_schema(&self.config, schema, self.profiles)?; // ── Fix B: Schema fingerprint persistence ─────────────────── // Persistent mode only. Compute a BLAKE3 hash of the schema's diff --git a/tidal/src/db/builder_tests.rs b/tidal/src/db/builder_tests.rs index e3e1517..c1ed775 100644 --- a/tidal/src/db/builder_tests.rs +++ b/tidal/src/db/builder_tests.rs @@ -28,6 +28,7 @@ fn builder_persistent_requires_data_dir() { schema: None, rate_limiter_config: None, transport: None, + profiles: Vec::new(), }; let result = builder.validate(); assert!(result.is_err()); diff --git a/tidal/src/db/open.rs b/tidal/src/db/open.rs index e9dab8f..31b285d 100644 --- a/tidal/src/db/open.rs +++ b/tidal/src/db/open.rs @@ -9,6 +9,7 @@ use crate::entities::{ CreatorItemsBitmap, HardNegIndex, InteractionLedger, PreferenceVectors, UserStateIndex, }; use crate::ranking::builtins::register_builtins; +use crate::ranking::profile::RankingProfile; use crate::ranking::registry::ProfileRegistry; use crate::schema::{DurabilityError, EntityId, Schema, TidalError, Timestamp}; use crate::signals::{NoopWalWriter, SignalLedger, SignalTypeId}; @@ -61,6 +62,7 @@ impl super::TidalDb { pub(crate) fn open_with_schema( config: &super::Config, schema: Schema, + schema_profiles: Vec, ) -> crate::Result { let last_seq = Arc::new(AtomicU64::new(0)); @@ -70,6 +72,17 @@ impl super::TidalDb { TidalError::internal("open", format!("failed to register builtin profiles: {e}")) })?; + // Register schema-defined profiles, overriding matching builtins. + for profile in schema_profiles { + let name = profile.name.clone(); + profile_registry.override_register(profile).map_err(|e| { + TidalError::internal( + "open", + format!("failed to register schema profile '{name}': {e}"), + ) + })?; + } + // Initialize M2 indexes (empty -- populated as items are written). let category_index = BitmapIndex::new("category"); let format_index = BitmapIndex::new("format"); diff --git a/tidal/src/ranking/registry.rs b/tidal/src/ranking/registry.rs index 49a8a84..b4b36f4 100644 --- a/tidal/src/ranking/registry.rs +++ b/tidal/src/ranking/registry.rs @@ -115,6 +115,46 @@ impl ProfileRegistry { Ok(()) } + /// Override-register a profile, replacing all existing versions for that name. + /// + /// Clears any previously registered versions (including builtins) and inserts + /// the new profile as the sole version. Applies the same validation as + /// [`register`](Self::register) (name format, exploration bounds, gate + /// thresholds) but skips version monotonicity since all prior versions are + /// removed. + /// + /// This is used by schema-defined profiles to cleanly replace built-in + /// defaults without version gymnastics. + /// + /// # Errors + /// + /// Same as [`register`](Self::register) except `VersionConflict` is impossible. + pub fn override_register(&mut self, profile: RankingProfile) -> Result<(), ProfileError> { + if !is_valid_name(&profile.name) { + return Err(ProfileError::InvalidName(profile.name)); + } + + if !(0.0..=0.5).contains(&profile.exploration) { + return Err(ProfileError::ExplorationOutOfRange(profile.exploration)); + } + + for gate in &profile.gates { + if matches!(gate.agg, SignalAgg::DecayScore | SignalAgg::Ratio) { + if !(0.0..=1.0).contains(&gate.min_threshold) { + return Err(ProfileError::GateThresholdOutOfRange(gate.min_threshold)); + } + } else if gate.min_threshold < 0.0 { + return Err(ProfileError::GateThresholdOutOfRange(gate.min_threshold)); + } + } + + // Clear all existing versions for this name, then insert the new one. + let versions = self.profiles.entry(profile.name.clone()).or_default(); + versions.clear(); + versions.insert(profile.version, profile); + Ok(()) + } + /// Get the latest version of a named profile. /// /// # Errors @@ -289,4 +329,68 @@ mod tests { let result = registry.get("nonexistent"); assert!(matches!(result, Err(ProfileError::NotFound(_)))); } + + #[test] + fn override_register_replaces_builtin() { + let mut registry = ProfileRegistry::new(); + let mut builtin = minimal_profile("for_you", 1); + builtin.is_builtin = true; + registry.register(builtin).unwrap(); + + // Override with a schema-defined version. + let mut override_profile = minimal_profile("for_you", 1); + override_profile.exploration = 0.1; + registry.override_register(override_profile).unwrap(); + + let profile = registry.get("for_you").unwrap(); + assert!(!profile.is_builtin); + assert!((profile.exploration - 0.1).abs() < f64::EPSILON); + } + + #[test] + fn override_register_clears_all_versions() { + let mut registry = ProfileRegistry::new(); + registry.register(minimal_profile("trending", 1)).unwrap(); + registry.register(minimal_profile("trending", 2)).unwrap(); + + // Override clears both v1 and v2. + let override_profile = minimal_profile("trending", 1); + registry.override_register(override_profile).unwrap(); + + // Only v1 exists now. + assert!(registry.get_version("trending", 2).is_err()); + assert!(registry.get_version("trending", 1).is_ok()); + } + + #[test] + fn override_register_validates_constraints() { + let mut registry = ProfileRegistry::new(); + + // Invalid name. + let result = registry.override_register(minimal_profile("BAD NAME", 1)); + assert!(matches!(result, Err(ProfileError::InvalidName(_)))); + + // Exploration out of range. + let mut profile = minimal_profile("bad_explore", 1); + profile.exploration = 0.9; + let result = registry.override_register(profile); + assert!(matches!( + result, + Err(ProfileError::ExplorationOutOfRange(_)) + )); + + // Gate threshold out of range. + let mut profile = minimal_profile("bad_gate", 1); + profile.gates.push(Gate { + signal: "view".into(), + agg: SignalAgg::DecayScore, + window: Window::AllTime, + min_threshold: 1.5, + }); + let result = registry.override_register(profile); + assert!(matches!( + result, + Err(ProfileError::GateThresholdOutOfRange(_)) + )); + } }