jordan/tidaldb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' of github.com:orchard9/tidalDB

Browse Source
This commit is contained in:
jordan 2026-03-04 17:13:36 -07:00
commit c4e0e49d6a
14 changed files with 1006 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -3325,6 +3325,7 @@ dependencies = [
"serde_json",
"serde_yaml",
"subtle",
"tempfile",
"thiserror 2.0.18",
"tidaldb",
"tokio",

32
docker/deploy/Dockerfile Normal file
View File

@ -0,0 +1,32 @@
FROM rust:1.91-slim AS builder
WORKDIR /build
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 ./
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:${PORT:-9500}/health || exit 1
ENV TIDAL_SERVER_LOG=info
ENV PORT=9500
ENTRYPOINT ["/usr/local/bin/tidal-server"]
CMD ["standalone", "--schema", "/config/schema.yaml", "--data-dir", "/data"]

5
tidal-server/Cargo.toml
View File

@ -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"] }
@ -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"

631
tidal-server/src/config.rs
View File

@ -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<Schema> {
pub fn load_schema(path: Option<&Path>) -> Result<(Schema, Vec<RankingProfile>)> {
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<Schema> {
));
}
// Collect signal names for profile validation.
let signal_names: Vec<String> = 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<Vec<Window>> = windows.iter().map(|w| parse_window(w)).collect();
sig = sig.windows(&parsed?);
}
@ -37,13 +44,13 @@ pub fn load_schema(path: Option<&Path>) -> Result<Schema> {
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<Schema> {
}
}
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<String> {
@ -70,6 +92,8 @@ struct SchemaSpec {
text_fields: Option<Vec<TextFieldSpec>>,
#[serde(default)]
embedding_slots: Option<Vec<EmbeddingSpec>>,
#[serde(default)]
profiles: Option<Vec<ProfileSpec>>,
}
#[derive(Debug, Deserialize)]
@ -171,3 +195,598 @@ fn parse_text_field_type(input: &str) -> Result<TextFieldType> {
))),
}
}
// ── Profile YAML types ──────────────────────────────────────────────────────
#[derive(Debug, Deserialize)]
struct ProfileSpec {
name: String,
#[serde(default = "default_version")]
version: u32,
#[serde(default)]
candidate_strategy: Option<CandidateStrategySpec>,
#[serde(default)]
sort: Option<SortSpec>,
#[serde(default)]
boosts: Vec<BoostSpec>,
#[serde(default)]
decay: Option<ProfileDecaySpec>,
#[serde(default)]
gates: Vec<GateSpec>,
#[serde(default)]
penalties: Vec<PenaltySpec>,
#[serde(default)]
excludes: Vec<ExcludeSpec>,
#[serde(default)]
diversity: Option<DiversitySpecConfig>,
#[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<usize>,
#[serde(default)]
format_mix_max_fraction: Option<f64>,
}
/// 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<String, serde_yaml::Value>),
}
/// 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<String, serde_yaml::Value>),
}
// ── Profile conversion ──────────────────────────────────────────────────────
impl ProfileSpec {
fn to_ranking_profile(&self) -> Result<RankingProfile> {
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::<Result<Vec<_>>>()?;
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::<Result<Vec<_>>>()?;
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::<Result<Vec<_>>>()?;
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::<Result<Vec<_>>>()?;
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<SignalAgg> {
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<Sort> {
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<CandidateStrategy> {
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<Option<f64>> {
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<Option<u64>> {
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<String> {
extract_string_field(val, field)
}
fn extract_string_field(val: &serde_yaml::Value, field: &str) -> Result<String> {
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<Window> {
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 { .. }
));
}
}

35
tidal-server/src/main.rs
View File

@ -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<PathBuf>,
@ -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<String, String> {
if s.contains(':') {
s.parse::<SocketAddr>()
.map(|_| s.to_string())
.map_err(|e| format!("invalid address '{s}': {e}"))
} else {
s.parse::<u16>()
.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()
@ -62,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 {
@ -111,13 +127,16 @@ async fn serve(state: ServerState, addr: &str, api_key: Option<Arc<str>>) -> 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<ServerState>) {
// SIGTERM is Unix-only; on other platforms we fall back to ctrl-c alone.
#[cfg(unix)]
let sigterm = async {
@ -146,5 +165,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");
}

44
tidal-server/src/router.rs
View File

@ -75,6 +75,8 @@ pub fn build_router(state: Arc<ServerState>, api_key: Option<Arc<str>>) -> 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,44 @@ 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<serde_json::Value> {
Json(serde_json::json!({ "ok": true, "service": "tidaldb" }))
}
/// Liveness probe: always 200 (process is alive).
async fn health_live() -> Json<serde_json::Value> {
Json(serde_json::json!({ "ok": true, "service": "tidaldb" }))
}
/// Readiness probe: 200 when ready, 503 when shutting down.
async fn health(
State(state): State<Arc<ServerState>>,
Query(query): Query<HashMap<String, String>>,
) -> Result<Json<HealthResponse>, AppError> {
) -> std::result::Result<(StatusCode, Json<serde_json::Value>), 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!({
"ok": true,
"service": "tidaldb",
"mode": "standalone",
"items": items,
})),
))
}
struct TidalErrorWrapper(tidaldb::TidalError);

15
tidal-server/src/state.rs
View File

@ -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<TidalDb>,
shutting_down: Arc<AtomicBool>,
}
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(

3
tidal-server/tests/middleware.rs
View File

@ -17,10 +17,11 @@ use tower::ServiceExt;
fn make_state() -> Arc<ServerState> {
// 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))

18
tidal/src/db/builder.rs
View File

@ -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<std::sync::Arc<dyn crate::replication::Transport>>,
/// Schema-defined profiles that override matching builtins.
profiles: Vec<RankingProfile>,
}
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<RankingProfile>) -> 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

1
tidal/src/db/builder_tests.rs
View File

@ -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());

29
tidal/src/db/open.rs
View File

@ -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};
@ -22,7 +23,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.
///
@ -61,6 +62,7 @@ impl super::TidalDb {
pub(crate) fn open_with_schema(
config: &super::Config,
schema: Schema,
schema_profiles: Vec<RankingProfile>,
) -> crate::Result<OpenResult> {
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");
@ -177,6 +190,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()

92
tidal/src/db/state_rebuild.rs
View File

@ -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;
@ -268,6 +272,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<u32>,
created_at_index: &RangeIndex<u64>,
) -> 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::<u32>()
{
duration_index.insert(id_u32, dur);
}
if let Some(val) = meta.get("created_at")
&& let Ok(ts) = val.parse::<u64>()
{
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

104
tidal/src/ranking/registry.rs
View File

@ -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(_))
));
}
}

32
tidal/tests/m2_uat.rs
View File

@ -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)