Compare commits
4 Commits
2dc00538e8
...
6ad8c51cfa
| Author | SHA1 | Date | |
|---|---|---|---|
|
6ad8c51cfa | ||
|
|
936da3c520 | ||
|
|
be37c72b12 | ||
|
|
34cc5f08ca |
@ -1,31 +1,31 @@
|
||||
version: 1
|
||||
project: tidalDB
|
||||
active_features:
|
||||
- m10-agent-capability-boundaries
|
||||
- m10-community-policy-engine
|
||||
- m10-signal-revocation-controls
|
||||
- m9-community-profile-sync
|
||||
- m9-leave-revocation
|
||||
- m9-retroactive-purge
|
||||
- m9-purge-rematerialization
|
||||
- m10-community-policy-engine
|
||||
- m10-agent-capability-boundaries
|
||||
- m10-signal-revocation-controls
|
||||
- p0-target-segment-recruitment
|
||||
- m9-retroactive-purge
|
||||
- p0-concierge-pilot-loop
|
||||
- p0-target-segment-recruitment
|
||||
- p0-validation-readout
|
||||
- p1-briefing-ux-reason-labels
|
||||
- p1-feedback-loop-ux
|
||||
- p1-quality-diversity-baseline
|
||||
- pg1-personalization-correctness
|
||||
- pg1-baseline-comparison
|
||||
- pg1-instrumented-metrics
|
||||
- p2-self-serve-onboarding
|
||||
- p2-cohort-context-views
|
||||
- p2-self-serve-onboarding
|
||||
- p2-trust-controls
|
||||
- p3-reliability-slos
|
||||
- p3-quality-operations
|
||||
- p3-launch-support-playbook
|
||||
- p3-quality-operations
|
||||
- p3-reliability-slos
|
||||
- p4-monetization-experiments
|
||||
- p4-quality-safe-growth
|
||||
- p4-segment-expansion-plan
|
||||
- pg1-baseline-comparison
|
||||
- pg1-instrumented-metrics
|
||||
- pg1-personalization-correctness
|
||||
active_directives: []
|
||||
history:
|
||||
- feature: m10-community-policy-engine
|
||||
@ -197,6 +197,7 @@ blocked: []
|
||||
milestones:
|
||||
- m0
|
||||
- m1
|
||||
- m10
|
||||
- m2
|
||||
- m3
|
||||
- m4
|
||||
@ -205,12 +206,11 @@ milestones:
|
||||
- m7
|
||||
- m8
|
||||
- m9
|
||||
- m10
|
||||
- p0
|
||||
- p1
|
||||
- pg1
|
||||
- p2
|
||||
- p3
|
||||
- p4
|
||||
- pg1
|
||||
active_ponders: []
|
||||
last_updated: 2026-03-05T00:38:49.555542Z
|
||||
last_updated: 2026-09-03T11:16:39.869277Z
|
||||
|
||||
53
API.md
53
API.md
@ -603,7 +603,7 @@ Sort modes are embedded in ranking profiles. The application names a profile. Th
|
||||
|---|---|
|
||||
| `new` | `created_at` DESC |
|
||||
| `trending` | Engagement velocity |
|
||||
| `hot` | Score / (age + 2)^gravity |
|
||||
| `hot` | `log10(max(views, 1)) / (age_hours + 2)^gravity`; 0/1 views tie |
|
||||
| `top_week` / `top_month` / `top_all_time` | Cumulative quality by window |
|
||||
| `most_viewed` / `most_liked` | Signal count by window |
|
||||
| `most_commented` / `most_shared` | Signal count (AllTime) |
|
||||
@ -852,7 +852,7 @@ db.reload_text_index()?;
|
||||
| **Scope trending by cohort** | Specify cohort name in retrieve query | Cohort-scoped signal aggregation, same ranking profile |
|
||||
| **Search within scope** | Specify `within` on search query | Intersects text/vector retrieval with scoped candidate set |
|
||||
| **HTTP write** | `POST /items`, `/embeddings`, `/signals` | Same as library write path, via JSON |
|
||||
| **HTTP query** | `GET /feed`, `/search` | Same as library query, via query params |
|
||||
| **HTTP query** | `GET /feed`, `/search`; standalone-only `POST /rank` | Stored-profile retrieval plus exact caller-supplied ranking |
|
||||
|
||||
One process. One query interface. One operational model.
|
||||
|
||||
@ -1089,6 +1089,49 @@ lies in `[0.0, 4.0]` and is monotonic with cosine distance), ordered closest-fir
|
||||
across a node's hosted shard groups in cluster mode). `unavailable_shards` is
|
||||
present only when a cross-shard probe was degraded (partial nearest-set).
|
||||
|
||||
#### `POST /rank` (standalone only)
|
||||
|
||||
Rank one complete caller-owned candidate snapshot without reading tidalDB item,
|
||||
signal, user, ledger, vector, or configured-profile state. The route is absent
|
||||
from cluster nodes until exact-profile version authority has a distributed
|
||||
contract.
|
||||
|
||||
```json
|
||||
{
|
||||
"profile": "contest_qualified_hot",
|
||||
"profile_version": 1,
|
||||
"as_of_nanos": "2000000000000000000",
|
||||
"candidates": [
|
||||
{
|
||||
"entity_id": 1,
|
||||
"created_at_nanos": "1999913600000000000",
|
||||
"signals": {"hearts": 8, "comments": 4, "exposure": 100}
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
`as_of_nanos` and every `created_at_nanos` are required decimal strings so JSON
|
||||
does not truncate a `u64`. Version 1 accepts at most 10,000 unique request-local
|
||||
entity IDs, rejects future creation times and negative/non-finite signals, and
|
||||
returns every candidate exactly once. Its deterministic order is
|
||||
`score DESC, entity_id DESC`.
|
||||
|
||||
The successful response repeats the profile identity and pinned clock, then
|
||||
returns `rank`, `score`, and the seven auditable components `hearts`,
|
||||
`comments`, `exposure`, `engagement`, `response_rate`, `age_hours`, and
|
||||
`freshness` for every item. Version 1 freezes this arithmetic:
|
||||
|
||||
```text
|
||||
engagement = hearts + 0.25 * comments
|
||||
response_rate = (engagement + 2) / (max(exposure, engagement) + 20)
|
||||
age_hours = (as_of_nanos - created_at_nanos) / 3_600_000_000_000
|
||||
freshness = (48 / (age_hours + 48))^0.25
|
||||
score = ln(1 + engagement) * freshness
|
||||
```
|
||||
|
||||
`response_rate` is diagnostic only; exposure never changes the v1 score.
|
||||
|
||||
### Health Endpoints
|
||||
|
||||
| Endpoint | Auth | Description |
|
||||
@ -1096,7 +1139,7 @@ present only when a cross-shard probe was degraded (partial nearest-set).
|
||||
| `GET /health` | No | Readiness probe — 200 when ready, 503 when shutting down |
|
||||
| `GET /health/startup` | No | Startup probe — always 200 |
|
||||
| `GET /health/live` | No | Liveness probe — always 200 |
|
||||
| `GET /openapi.json` | No | Machine-readable OpenAPI 3.1 spec for this server (data + cluster routes) |
|
||||
| `GET /openapi.json` | No | Machine-readable contract for the current binary; standalone omits cluster routes and cluster omits `/rank` |
|
||||
|
||||
These map directly to Kubernetes startup/liveness/readiness probes — see
|
||||
[docs/runbooks/kubernetes.md](docs/runbooks/kubernetes.md).
|
||||
@ -1110,8 +1153,8 @@ OpenAPI viewer or generate a client:
|
||||
curl -s http://localhost:9400/openapi.json | jq '.info, (.paths | keys)'
|
||||
```
|
||||
|
||||
The cluster server serves a superset document that also includes the
|
||||
`/cluster/*` and `/sharded/*` routes. See
|
||||
The cluster document adds `/cluster/*` and `/sharded/*` routes but intentionally
|
||||
omits standalone-only `POST /rank`. See
|
||||
[docs/guides/server-deployment.md](docs/guides/server-deployment.md).
|
||||
|
||||
### Cluster Endpoints
|
||||
|
||||
@ -4,6 +4,15 @@ All notable changes to tidalDB will be documented in this file.
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
### Added
|
||||
|
||||
- Standalone `POST /rank` and `TidalDb::rank_exact` provide bounded,
|
||||
deterministic exact ranking over a complete caller-supplied candidate set.
|
||||
The first frozen profile is `contest_qualified_hot` version 1; responses
|
||||
include the pinned clock, every candidate exactly once, and auditable score
|
||||
components. Cluster routers intentionally omit this route until exact-profile
|
||||
version authority has a distributed contract.
|
||||
|
||||
### Stability
|
||||
|
||||
tidalDB is **production-ready**. M0–M12 are shipped and the pre-release
|
||||
|
||||
@ -226,7 +226,7 @@ for item in &results.items {
|
||||
```
|
||||
|
||||
Other useful profiles:
|
||||
- `"hot"` — score with age decay (Reddit model)
|
||||
- `"hot"` — cumulative view count with explicit age decay
|
||||
- `"following"` — content from followed creators (requires `for_user` + written `follows` relationships)
|
||||
- `"hidden_gems"` — high completion rate, low reach
|
||||
- `"top_week"` — cumulative quality over the last 7 days
|
||||
|
||||
20
USE_CASES.md
20
USE_CASES.md
@ -283,7 +283,7 @@ This is the Reddit "rising" concept applied broadly.
|
||||
|
||||
**New** — pure reverse chronological. No quality gate. Shows everything. Users use this to find content the algorithm hasn't surfaced yet.
|
||||
|
||||
**Hot** — recency + engagement combined. Content decays as it ages regardless of engagement. The Reddit model: score / (age_hours + 2)^gravity. Refreshes meaningfully every hour.
|
||||
**Hot** — cumulative all-time views with explicit age decay: `log10(max(views, 1)) / (age_hours + 2)^gravity`. Age affects the order only from the second view onward; zero-view and one-view items both score zero.
|
||||
|
||||
**Rising** — overperforming new content (see UC-03.2).
|
||||
|
||||
@ -539,15 +539,21 @@ Controversial is defined as: high total engagement AND polarized sentiment. High
|
||||
|
||||
**Ranking Profile:** `controversial` — maximizes the product of positive and negative engagement signals. A post with 1000 upvotes and 1000 downvotes scores higher than one with 1800 upvotes and 200 downvotes.
|
||||
|
||||
### 14.2 · Hot Sort (Reddit Model)
|
||||
### 14.2 · Hot Sort (View-Count Decay)
|
||||
|
||||
**Surface:** Reddit "Hot," Hacker News front page, time-sensitive community surfaces.
|
||||
**Surface:** Time-sensitive, popularity-weighted content surfaces.
|
||||
|
||||
**The Question:** What is the best content right now, with age decay applied?
|
||||
**The Question:** What viewed content is popular enough to survive explicit age decay?
|
||||
|
||||
Hot rewards early engagement but punishes age. Formula concept: `score / (age_hours + 2)^gravity`. The database exposes this as a native sort mode — the application does not implement the formula.
|
||||
Hot uses the cumulative all-time `view` count and item age:
|
||||
`log10(max(views, 1)) / (age_hours + 2)^gravity`. The database exposes this as
|
||||
a native sort mode; applications do not substitute votes or implement the
|
||||
formula themselves. The numerator is zero at both zero and one view, so age
|
||||
cannot break ties in that cold-start band.
|
||||
|
||||
**What makes Hot different from Trending:** Trending is pure velocity (rate of change). Hot is cumulative score with age decay. An hour-old post with 500 upvotes scores higher on Hot than a day-old post with 2000 upvotes.
|
||||
**What makes Hot different from Trending:** Trending is recent velocity. Hot is
|
||||
cumulative view count with age decay, so equally old items order by views while
|
||||
equally viewed items with at least two views order newest first.
|
||||
|
||||
---
|
||||
|
||||
@ -713,7 +719,7 @@ All sort modes must be available on any surface. The application specifies the s
|
||||
| `top_week` | Quality score, last 7d | Weekly digest |
|
||||
| `top_month` | Quality score, last 30d | Monthly recap |
|
||||
| `top_year` | Quality score, last 365d | Annual best |
|
||||
| `hot` | Score / (age + 2)^gravity — decays with time | Community frontpages |
|
||||
| `hot` | `log10(max(views, 1)) / (age_hours + 2)^gravity`; 0/1 views tie | Community frontpages |
|
||||
| `trending` | Pure engagement velocity | Trending tabs |
|
||||
| `rising` | Velocity relative to baseline, age-boosted | Breakout content |
|
||||
| `controversial` | max(positive_signals × negative_signals) | Debate/discussion |
|
||||
|
||||
@ -111,7 +111,7 @@ tidalDB is designed to handle every retrieval and ranking pattern a content plat
|
||||
- Relevance (text + semantic match)
|
||||
- Personalized (user preference match)
|
||||
- New / Old (chronological)
|
||||
- Hot (score with age decay — Reddit model)
|
||||
- Hot (cumulative view count with explicit age decay)
|
||||
- Trending (pure velocity)
|
||||
- Rising (velocity relative to creator/category baseline, age-boosted)
|
||||
- Top: All Time / This Year / This Month / This Week / Today / This Hour
|
||||
|
||||
@ -11,7 +11,7 @@
|
||||
- Sort modes are built-in, not formulas the application implements
|
||||
- Same sort mode works across different candidate sets (global, category, social graph)
|
||||
- Windowed top sorts: hour, today, week, month, year, all-time
|
||||
- Hot uses Reddit-style age decay: score / (age + 2)^gravity
|
||||
- Hot uses `log10(max(views, 1)) / (age_hours + 2)^gravity`; zero-view and one-view items tie because both numerators are zero
|
||||
- Trending is pure velocity (rate of change), distinct from Hot (cumulative with decay)
|
||||
- Controversial maximizes product of positive and negative signals
|
||||
|
||||
|
||||
@ -23,7 +23,7 @@ A ranking profile is a named, versioned bundle that fully specifies how a query
|
||||
| Profile | Optimizes for | Primary sort | Diversity / exploration (defaults) | Needs FOR USER | Use case |
|
||||
|---------|---------------|--------------|------------------------------------|----------------|----------|
|
||||
| `trending` | Pure short-window engagement velocity (`view_vel + 2·share_vel`, 24h) | `Trending` | `max_per_creator=1`; no exploration | No | UC-03 |
|
||||
| `hot` | Cumulative score decayed by age (Reddit/HN style) | `Hot { gravity = 1.8 }` | `max_per_creator=2`; no exploration | No | UC-14 |
|
||||
| `hot` | Cumulative all-time views with explicit age decay | `Hot { gravity = 1.8 }` | `max_per_creator=2`; no exploration | No | UC-14 |
|
||||
| `new` | Freshest content first | `New` (`created_at DESC`) | none | No | UC-03, UC-04 |
|
||||
| `for_you` | Personalized home feed (decayed view/like + 24h share velocity, on top of preference match) | `Hot { gravity = 1.5 }` | `max_per_creator=2`, `format_mix_max_fraction=0.4`, `exploration=0.1` | Yes | UC-01 |
|
||||
| `following` | Recent content from creators the user follows | `New` | `max_per_creator=3`; no exploration | Yes | UC-04 |
|
||||
|
||||
@ -322,7 +322,7 @@ curl -s localhost:9400/openapi.json | jq .info
|
||||
# }
|
||||
```
|
||||
|
||||
You can load `/openapi.json` into any OpenAPI viewer (Swagger UI, Redoc, Stoplight, Postman) or feed it to a client generator (`openapi-generator`, `oapi-codegen`, etc.) to produce typed clients in any language. The standalone document describes the data + health surface; the cluster document is a superset that also describes the `/cluster/*` and `/sharded/*` routes. The `bearerAuth` security scheme is declared on the data routes so generated clients know to send the token; `/openapi.json` and the probes carry no security requirement.
|
||||
You can load `/openapi.json` into any OpenAPI viewer (Swagger UI, Redoc, Stoplight, Postman) or feed it to a client generator (`openapi-generator`, `oapi-codegen`, etc.) to produce typed clients in any language. The standalone document describes the data + health surface. The cluster document adds `/cluster/*` and `/sharded/*`, but intentionally omits standalone-only `POST /rank` until exact-profile version authority has a distributed contract. The `bearerAuth` security scheme is declared on the data routes so generated clients know to send the token; `/openapi.json` and the probes carry no security requirement.
|
||||
|
||||
### Route summary (standalone)
|
||||
|
||||
@ -338,8 +338,9 @@ You can load `/openapi.json` into any OpenAPI viewer (Swagger UI, Redoc, Stoplig
|
||||
| `GET /feed` | yes | `200` | `?user_id=&profile=for_you&limit=20`. |
|
||||
| `GET /search` | yes | `200` | `?query=&user_id=&limit=20`. |
|
||||
| `POST /vector_search` | yes | `200` | Pure k-NN. Body: `{ "vector": [<f32>, ...], "k"?, "ef_search"? }`. |
|
||||
| `POST /rank` | yes | `200` | Standalone-only exact ranking over a complete caller-supplied candidate set; the profile, version, and decimal-string `as_of_nanos` are required. |
|
||||
|
||||
`limit` is clamped to `1000` at the trust boundary. Request middleware: 30 s timeout (`408`), 100 max in-flight (`429`), 2 MB body cap (`413`), and an `x-request-id` echoed on every response.
|
||||
Feed/search `limit` is clamped to `1000`; exact rank accepts at most 10,000 candidates and returns each exactly once. Request middleware: 30 s timeout (`408`), 100 max in-flight (`429`), 2 MB body cap (`413`), and an `x-request-id` echoed on every response.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@ -551,7 +551,7 @@ Given:
|
||||
- Ranking profiles defined:
|
||||
* "trending" -- share_velocity(6h) primary, view_velocity(6h) secondary,
|
||||
engagement_ratio gate > 0.03
|
||||
* "hot" -- score / (age_hours + 2)^1.8
|
||||
* "hot" -- log10(max(views, 1)) / (age_hours + 2)^1.8
|
||||
* "new" -- created_at DESC
|
||||
* "top_week" -- quality_score within 7d window
|
||||
* "hidden_gems" -- high completion_rate, inverse view_count
|
||||
@ -633,7 +633,7 @@ Then:
|
||||
- [x] Profiles stored in schema, versioned, retrievable by name
|
||||
- [x] Profile execution: given a candidate set and a profile, produce a scored and sorted result list
|
||||
- [x] Built-in profiles implemented: `trending`, `hot`, `new`, `top_week`, `top_month`, `top_all_time`, `hidden_gems`, `controversial`, `most_viewed`, `most_liked`, `shuffle`
|
||||
- [x] `hot` formula: `log10(max(|positive - negative|, 1)) / (age_hours + 2)^gravity` with configurable gravity
|
||||
- [x] `hot` formula: `log10(max(views, 1)) / (age_hours + 2)^gravity` with configurable gravity
|
||||
- [x] `controversial` formula: `(positive * negative) / (positive + negative)^2`
|
||||
- [x] `hidden_gems` formula: `quality_score * (1 / log10(view_count + 10))` -- the `+10` prevents division by zero for items with zero views
|
||||
- [x] Profile change does not require recompile -- profiles are runtime data
|
||||
|
||||
@ -18,7 +18,7 @@ This is the phase that turns signals from "primitives the application reads" int
|
||||
- [ ] Built-in profiles registered at `SchemaBuilder::build()` time: `trending`, `hot`, `new`, `top_week`, `top_month`, `top_all_time`, `hidden_gems`, `controversial`, `most_viewed`, `most_liked`, `shuffle`
|
||||
- [ ] Built-in profiles are standard `RankingProfile` instances -- not special-cased in the executor
|
||||
- [ ] Built-in profiles with unavailable signals degrade gracefully (skip missing signals, not fatal error)
|
||||
- [ ] `hot` formula: `log10(max(|positive - negative|, 1)) / (age_hours + 2)^gravity` with configurable gravity (default 1.8) -- Spec 09 Section 11.1
|
||||
- [ ] `hot` formula: `log10(max(views, 1)) / (age_hours + 2)^gravity` with configurable gravity (default 1.8) -- Spec 09 Section 11.1
|
||||
- [ ] `controversial` formula: `(positive * negative) / (positive + negative)^2` -- Spec 09 Section 11.4
|
||||
- [ ] `hidden_gems` formula: `quality_score * (1 / log10(view_count + 10))` -- Spec 09 Section 11.5
|
||||
- [ ] `ProfileExecutor::score()` takes `&[EntityId]` candidates and `&RankingProfile`, returns `Vec<ScoredCandidate>` sorted by score descending
|
||||
|
||||
@ -397,7 +397,7 @@ pub struct DiversitySpec {
|
||||
/// normalization, and diversity still apply.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub enum Sort {
|
||||
/// `log10(max(|positive - negative|, 1)) / (age_hours + 2)^gravity`
|
||||
/// `log10(max(views, 1)) / (age_hours + 2)^gravity`
|
||||
/// Spec 09 Section 11.1. Default gravity: 1.8.
|
||||
Hot { gravity: f64 },
|
||||
|
||||
|
||||
@ -154,9 +154,10 @@ fn builtin_trending() -> RankingProfile {
|
||||
p
|
||||
}
|
||||
|
||||
/// hot: score / (age_hours + 2)^gravity. Spec 09 Section 13.10.
|
||||
/// hot: log10(max(views, 1)) / (age_hours + 2)^gravity.
|
||||
/// Spec 09 Section 13.10.
|
||||
///
|
||||
/// Requires: like, dislike (for positive/negative computation)
|
||||
/// Requires: view (all-time aggregate) and item created_at metadata.
|
||||
/// Sort formula replaces boost/penalty pipeline.
|
||||
fn builtin_hot() -> RankingProfile {
|
||||
let mut p = RankingProfile::new("hot", 1);
|
||||
@ -325,7 +326,7 @@ fn builtin_shuffle() -> RankingProfile {
|
||||
| Profile | Required Signals | Required Windows | Requires Velocity |
|
||||
|---------|-----------------|------------------|-------------------|
|
||||
| `trending` | share, view | 1h, 24h | Yes (share, view) |
|
||||
| `hot` | like, dislike | all_time | No |
|
||||
| `hot` | view | all_time | No |
|
||||
| `new` | (none) | (none) | No |
|
||||
| `top_week` | view, like, share, completion | 7d | No |
|
||||
| `top_month` | view, like, share, completion | 30d | No |
|
||||
|
||||
@ -13,7 +13,7 @@
|
||||
Deliver the `ProfileExecutor` that takes a `&RankingProfile` and a `&[EntityId]` of candidates, reads signal state from the `SignalLedger`, applies the profile's scoring rules (boosts, penalties, gates, sort formulas), and returns `Vec<ScoredCandidate>` sorted by score descending. This is the heart of tidalDB's ranking engine -- the function that turns "here are 200 candidate items" into "here are 200 items ranked by this profile."
|
||||
|
||||
The executor implements all sort mode formulas from Spec 09 Section 11:
|
||||
- **Hot:** `log10(max(|positive - negative|, 1)) / (age_hours + 2)^gravity`
|
||||
- **Hot:** `log10(max(views, 1)) / (age_hours + 2)^gravity`
|
||||
- **Controversial:** `(positive * negative) / (positive + negative)^2`
|
||||
- **Hidden Gems:** `quality_score * (1 / log10(view_count + 10))`
|
||||
- **Trending:** `share_velocity * 0.5 + view_velocity * 0.3 + reach_value * 0.2`
|
||||
@ -189,15 +189,13 @@ Each sort formula is a standalone function for testability:
|
||||
```rust
|
||||
// === ranking/executor.rs (internal functions) ===
|
||||
|
||||
/// Hot formula: log10(max(|positive - negative|, 1)) / (age_hours + 2)^gravity
|
||||
/// Hot formula: log10(max(views, 1)) / (age_hours + 2)^gravity
|
||||
///
|
||||
/// Spec 09 Section 11.1.
|
||||
/// positive = like.count(all_time)
|
||||
/// negative = dislike.count(all_time)
|
||||
/// views = view.count(all_time)
|
||||
/// age_hours = (now - created_at).as_secs_f64() / 3600.0
|
||||
fn hot_score(positive: u64, negative: u64, age_hours: f64, gravity: f64) -> f64 {
|
||||
let diff = (positive as f64 - negative as f64).abs().max(1.0);
|
||||
diff.log10() / (age_hours + 2.0).powf(gravity)
|
||||
fn hot_score(views: u64, age_hours: f64, gravity: f64) -> f64 {
|
||||
(views as f64).max(1.0).log10() / (age_hours + 2.0).powf(gravity)
|
||||
}
|
||||
|
||||
/// Controversial formula: (positive * negative) / (positive + negative)^2
|
||||
@ -628,18 +626,17 @@ criterion_main!(benches);
|
||||
|
||||
#[test]
|
||||
fn hot_score_basic() {
|
||||
// 100 likes, 10 dislikes, 1 hour old, gravity 1.8
|
||||
let score = hot_score(100, 10, 1.0, 1.8);
|
||||
// log10(|100-10|) / (1+2)^1.8 = log10(90) / 3^1.8
|
||||
// 90 views, 1 hour old, gravity 1.8
|
||||
let score = hot_score(90, 1.0, 1.8);
|
||||
// log10(90) / (1+2)^1.8
|
||||
let expected = 90.0_f64.log10() / 3.0_f64.powf(1.8);
|
||||
assert!((score - expected).abs() < 1e-10,
|
||||
"hot_score={score}, expected={expected}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hot_score_zero_engagement() {
|
||||
// No likes or dislikes -- score uses max(1, |0-0|)
|
||||
let score = hot_score(0, 0, 1.0, 1.8);
|
||||
fn hot_score_zero_views() {
|
||||
let score = hot_score(0, 1.0, 1.8);
|
||||
let expected = 1.0_f64.log10() / 3.0_f64.powf(1.8);
|
||||
assert!((score - expected).abs() < 1e-10);
|
||||
assert!((score - 0.0).abs() < 1e-10, "log10(1) = 0, so score should be 0");
|
||||
@ -647,18 +644,18 @@ fn hot_score_zero_engagement() {
|
||||
|
||||
#[test]
|
||||
fn hot_score_higher_gravity_lower_score() {
|
||||
let score_low = hot_score(100, 10, 6.0, 1.0);
|
||||
let score_high = hot_score(100, 10, 6.0, 2.5);
|
||||
let score_low = hot_score(90, 6.0, 1.0);
|
||||
let score_high = hot_score(90, 6.0, 2.5);
|
||||
assert!(score_low > score_high,
|
||||
"higher gravity should produce lower score for same age");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hot_score_older_content_scores_lower() {
|
||||
let score_new = hot_score(100, 10, 1.0, 1.8);
|
||||
let score_old = hot_score(100, 10, 24.0, 1.8);
|
||||
let score_new = hot_score(90, 1.0, 1.8);
|
||||
let score_old = hot_score(90, 24.0, 1.8);
|
||||
assert!(score_new > score_old,
|
||||
"newer content should score higher with same engagement");
|
||||
"newer content should score higher with the same views");
|
||||
}
|
||||
|
||||
#[test]
|
||||
@ -1017,14 +1014,13 @@ proptest! {
|
||||
proptest! {
|
||||
#[test]
|
||||
fn hot_score_decreases_with_age(
|
||||
positive in 1u64..10000,
|
||||
negative in 0u64..10000,
|
||||
views in 1u64..10000,
|
||||
age1 in 0.1f64..100.0,
|
||||
age_delta in 0.1f64..100.0,
|
||||
gravity in 0.5f64..3.0,
|
||||
) {
|
||||
let score1 = hot_score(positive, negative, age1, gravity);
|
||||
let score2 = hot_score(positive, negative, age1 + age_delta, gravity);
|
||||
let score1 = hot_score(views, age1, gravity);
|
||||
let score2 = hot_score(views, age1 + age_delta, gravity);
|
||||
prop_assert!(score1 >= score2,
|
||||
"hot score should decrease with age: age={age1} score={score1}, age={} score={score2}",
|
||||
age1 + age_delta);
|
||||
@ -1054,7 +1050,7 @@ proptest! {
|
||||
- [ ] `ProfileExecutor::new(ledger)` borrows a `SignalLedger`
|
||||
- [ ] `ProfileExecutor::score()` takes candidates, profile, now, optional shuffle_seed; returns `Vec<ScoredCandidate>` sorted descending
|
||||
- [ ] Sort override detection: when `profile.has_sort_override()`, sort formula replaces boost/penalty pipeline
|
||||
- [ ] `hot_score()` implements `log10(max(|positive - negative|, 1)) / (age_hours + 2)^gravity` matching Spec 09 Section 11.1
|
||||
- [ ] `hot_score()` implements `log10(max(views, 1)) / (age_hours + 2)^gravity` matching Spec 09 Section 11.1
|
||||
- [ ] `controversial_score()` implements `(positive * negative) / (positive + negative)^2` matching Spec 09 Section 11.4
|
||||
- [ ] `hidden_gems_score()` implements `quality_score * (1 / log10(view_count + 10))` matching Spec 09 Section 11.5
|
||||
- [ ] `top_window_score()` implements weighted signal sum matching Spec 09 Section 11.7
|
||||
|
||||
@ -1875,7 +1875,7 @@ Sort modes (from API.md) are implemented as sort expressions in the scan candida
|
||||
| `Personalized` | User preference vector similarity | Cosine similarity | DESC |
|
||||
| `New` | Metadata field read | `created_at` | DESC |
|
||||
| `Old` | Metadata field read | `created_at` | ASC |
|
||||
| `Hot` | `score / (age_hours + 2)^1.8` | Composite of signal + timestamp | DESC |
|
||||
| `Hot` | `log10(max(views, 1)) / (age_hours + 2)^1.8` | `view.count(all_time)` + `created_at` | DESC |
|
||||
| `Trending` | Signal velocity read | `view.velocity(6h)` + `share.velocity(6h)` | DESC |
|
||||
| `Rising` | Velocity relative to baseline | `velocity / baseline` | DESC |
|
||||
| `TopAllTime` | Signal accumulator | `like.decay_score(all_time)` | DESC |
|
||||
|
||||
@ -1071,26 +1071,30 @@ Sort modes are formula-based ranking functions that bypass the boost/penalty sco
|
||||
### 11.1 Hot
|
||||
|
||||
```
|
||||
hot_score(item) = log10(max(|positive - negative|, 1))
|
||||
hot_score(item) = log10(max(views, 1))
|
||||
/ (age_hours + 2) ^ gravity
|
||||
|
||||
Where:
|
||||
positive = upvotes + likes
|
||||
negative = downvotes + dislikes
|
||||
views = view.count(all_time)
|
||||
age_hours = (now - created_at).as_hours()
|
||||
gravity = configurable, default 1.8
|
||||
```
|
||||
|
||||
**Behavior:** Hot rewards early engagement but punishes age. An hour-old post with 500 upvotes scores higher than a day-old post with 2,000 upvotes. The gravity parameter controls how aggressively age suppresses score. Higher gravity = faster decay.
|
||||
**Behavior:** Hot rewards cumulative views while explicitly suppressing age.
|
||||
At equal age, higher view counts rank above lower counts once either count is
|
||||
at least two. At equal views of two or more, newer content ranks higher;
|
||||
candidates with zero or one view all score `0.0`. It does not read upvotes,
|
||||
likes, downvotes, or dislikes. Higher gravity makes the age penalty steeper.
|
||||
|
||||
| Gravity | Behavior |
|
||||
|---------|----------|
|
||||
| 1.0 | Very slow decay. Content stays hot for days. |
|
||||
| 1.5 | Moderate decay. Content refreshes every ~6 hours. |
|
||||
| 1.8 | Standard (Reddit default). Content refreshes every ~3 hours. |
|
||||
| 2.5 | Aggressive decay. Content refreshes hourly. |
|
||||
| 1.0 | Slow age decay. |
|
||||
| 1.5 | Moderate age decay. |
|
||||
| 1.8 | Default built-in `hot` decay. |
|
||||
| 2.5 | Aggressive age decay. |
|
||||
|
||||
**Use cases:** UC-06 (Browse/Category), UC-14 (Hot Surfaces), any community frontpage.
|
||||
**Use cases:** UC-06 (Browse/Category), UC-14 (Hot Surfaces), and other
|
||||
view-popularity surfaces.
|
||||
|
||||
### 11.2 Trending
|
||||
|
||||
|
||||
@ -1405,7 +1405,7 @@ The following profiles are automatically available after entity and signal types
|
||||
| `for_you` | ANN over user preference vector, top_k=500 | preference match + engagement velocity | Personalized blend of semantic relevance and social proof |
|
||||
| `trending` | Scan all items | `view.velocity(6h) + share.velocity(6h)` | Pure signal velocity, no personalization |
|
||||
| `rising` | Scan all items | Relative velocity: `velocity(1h) / velocity(24h)`, age-boosted | Content accelerating relative to its baseline |
|
||||
| `hot` | Scan all items | `score / (age_hours + 2)^1.8` | Reddit-model age decay over cumulative engagement |
|
||||
| `hot` | Scan all items | `log10(max(views, 1)) / (age_hours + 2)^1.8` | Cumulative view count with age decay; zero-view and one-view items tie |
|
||||
| `following` | Relationship: `follows` | N/A | `created_at DESC` (pure chronological) |
|
||||
| `related` | ANN over anchor item embedding, top_k=200 | Semantic similarity + collaborative filtering | Most similar content to the anchor |
|
||||
| `browse` | Scan all items | `completion_rate * 0.4 + like_ratio * 0.3 + log(views) * 0.3` | Quality-weighted with reach tiebreaker |
|
||||
|
||||
@ -309,8 +309,9 @@ spec:
|
||||
#
|
||||
# 300m is what the tightest pinned node (k3s-agent-1, 355m free)
|
||||
# can actually reserve for a voter, and it is honest for a FIRST
|
||||
# CONSUMER's load - not for the gate. The limit below keeps the
|
||||
# measured burst reachable without reserving it.
|
||||
# CONSUMER's load - not for the gate. It also matches measured p95
|
||||
# (302/235/233m per pod). The limit below keeps the measured burst
|
||||
# reachable without reserving it.
|
||||
#
|
||||
# This is provisioned optimism with named detectors: if real write
|
||||
# volume approaches the knee, TidalDBClusterQuorumLag,
|
||||
@ -321,7 +322,10 @@ spec:
|
||||
cpu: "300m"
|
||||
# Baseline working set was ~3.6 GiB before load, and full placement
|
||||
# means every pod holds the WHOLE 1536-D corpus. This is a resident
|
||||
# footprint, not a gate artifact: it stays at 4 GiB.
|
||||
# footprint, not a gate artifact: it stays at 4 GiB. An 11d
|
||||
# observation put pods at 4.4-5.7Gi, so the 1Gi this file once
|
||||
# declared would make every replica a Burstable-over-request
|
||||
# eviction candidate.
|
||||
memory: 4Gi
|
||||
limits:
|
||||
# 3 cores / 7Gi, matching what production actually runs.
|
||||
@ -333,6 +337,14 @@ spec:
|
||||
# are the intended ones and this file is now the source of truth for
|
||||
# them; raise BOTH together or not at all.
|
||||
#
|
||||
# Why 3 and not 2 (m12 read-SLA fix): the cgroup cpu quota is what
|
||||
# the engine's available_parallelism() reads (SEARCH_GATE /
|
||||
# worker_threads sizing). At "2" a cross-shard search burst starved
|
||||
# the async reactor plus the election/heartbeat/apply loops - reads
|
||||
# hung to the 30s route timeout, and the starved control plane
|
||||
# churned elections into reseed self-exit. "3" leaves ~1 core for
|
||||
# kubelet/system on the 4-core nodes.
|
||||
#
|
||||
# Why a limit above the request at all: it keeps the measured
|
||||
# query/apply burst reachable on the tightest node without reserving
|
||||
# it. Note the ratio - a burstable pod whose neighbours are also
|
||||
@ -343,8 +355,9 @@ spec:
|
||||
cpu: "3"
|
||||
# Four independent OOMKills occurred at 3.97-4.00 GiB. Six GiB was
|
||||
# measured peak plus 50% headroom; production carries 7Gi and that is
|
||||
# what is declared here. The exact internal growth source still
|
||||
# requires heap/allocation profiling.
|
||||
# what is declared here - tidaldb-0 held 5751Mi, 94% of the old 6Gi
|
||||
# ceiling. The exact internal growth source still requires
|
||||
# heap/allocation profiling.
|
||||
memory: 7Gi
|
||||
securityContext:
|
||||
allowPrivilegeEscalation: false
|
||||
|
||||
@ -2050,12 +2050,27 @@ impl ShardReplica {
|
||||
signal: &str,
|
||||
entity: EntityId,
|
||||
weight: f64,
|
||||
user_id: Option<u64>,
|
||||
creator_id: Option<u64>,
|
||||
) -> Result<StagedSignal> {
|
||||
if !self.is_leader() {
|
||||
return Err(self.not_leader());
|
||||
}
|
||||
let db = self.db()?;
|
||||
db.signal_staged(signal, entity, weight, Timestamp::now())
|
||||
// Always the context-carrying stage: it degrades to the plain staged
|
||||
// write when both ids are `None`, so there is no second code path to
|
||||
// keep in step. Using `signal_staged` here is what silently discarded
|
||||
// `user_id`/`creator_id` on every clustered write while still
|
||||
// answering 204 — no hard negatives, no seen tracking, no interaction
|
||||
// weight, no preference vector, and no wire evidence of the loss.
|
||||
db.signal_with_context_staged(
|
||||
signal,
|
||||
entity,
|
||||
weight,
|
||||
Timestamp::now(),
|
||||
user_id,
|
||||
creator_id,
|
||||
)
|
||||
.map_err(ServerError::Tidal)
|
||||
}
|
||||
|
||||
@ -7486,6 +7501,8 @@ pub async fn write_signal(
|
||||
let signal = req.signal;
|
||||
let entity = EntityId::new(req.entity_id);
|
||||
let weight = req.weight;
|
||||
let user_id = req.user_id;
|
||||
let creator_id = req.creator_id;
|
||||
// Two-phase write (m11p1): STAGE on the write pool (microseconds; the
|
||||
// bounded queue keeps the 429 admission semantics), then COMPLETE — the
|
||||
// group-commit fsync wait — on the blocking pool, freeing the pool worker
|
||||
@ -7506,7 +7523,8 @@ pub async fn write_signal(
|
||||
let ticket = state
|
||||
.write_pool
|
||||
.submit(move || {
|
||||
let staged = state_for_job.stage_signal_local(&signal, entity, weight)?;
|
||||
let staged =
|
||||
state_for_job.stage_signal_local(&signal, entity, weight, user_id, creator_id)?;
|
||||
Ok(StagedWriteTicket::new(staged, Arc::clone(&state_for_job)))
|
||||
})
|
||||
.await
|
||||
@ -9513,6 +9531,8 @@ pub async fn sharded_write_signal(
|
||||
let entity = EntityId::new(req.entity_id);
|
||||
let signal = req.signal.clone();
|
||||
let weight = req.weight;
|
||||
let user_id = req.user_id;
|
||||
let creator_id = req.creator_id;
|
||||
sharded_write_route(
|
||||
&state,
|
||||
&headers,
|
||||
@ -9520,7 +9540,14 @@ pub async fn sharded_write_signal(
|
||||
"/sharded/signals",
|
||||
&req,
|
||||
move || {
|
||||
db.signal(&signal, entity, weight, Timestamp::now())
|
||||
db.signal_with_context(
|
||||
&signal,
|
||||
entity,
|
||||
weight,
|
||||
Timestamp::now(),
|
||||
user_id,
|
||||
creator_id,
|
||||
)
|
||||
.map_err(ServerError::Tidal)
|
||||
},
|
||||
StatusCode::NO_CONTENT,
|
||||
|
||||
@ -395,6 +395,30 @@ pub async fn write_embedding(
|
||||
Ok(StatusCode::NO_CONTENT)
|
||||
}
|
||||
|
||||
/// Refuse a signal whose originating context this router cannot honour.
|
||||
///
|
||||
/// `SimulatedCluster::write_signal` and `scatter_gather::sharded_write_signal`
|
||||
/// both apply `(signal, entity, weight)` and nothing else, so `user_id` /
|
||||
/// `creator_id` would be accepted and thrown away — no hard negatives, no seen
|
||||
/// tracking, no interaction weight, no preference vector, and a 204 claiming
|
||||
/// success. Fail closed instead, naming the route that does support it.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// `ServerError::BadRequest` when either context id is present.
|
||||
fn reject_unsupported_signal_context(req: &SignalRequest) -> Result<()> {
|
||||
if req.user_id.is_some() || req.creator_id.is_some() {
|
||||
return Err(ServerError::BadRequest(
|
||||
"single-process cluster mode cannot record signal context: \
|
||||
user_id/creator_id are unsupported on this router. Run the \
|
||||
multi-process cluster (`--region`, the deployed RF3 topology), \
|
||||
which applies context via signal_with_context_staged."
|
||||
.to_owned(),
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Record a signal on the leader region and eagerly ship it to followers.
|
||||
///
|
||||
/// Returns `204 No Content` once the signal is **durably applied on the leader**
|
||||
@ -421,6 +445,13 @@ pub async fn write_signal(
|
||||
State(state): State<Arc<ClusterState>>,
|
||||
Json(req): Json<SignalRequest>,
|
||||
) -> std::result::Result<StatusCode, ClusterAppError> {
|
||||
// The simulated relay behind single-process mode applies (signal, entity,
|
||||
// weight) only, so it CANNOT honour originating context. Refuse rather than
|
||||
// accept-and-discard: a 204 on a dropped `user_id` is the failure mode that
|
||||
// made a clustered deployment look healthy while learning nothing. The
|
||||
// production multi-process path (`build_region_router`) carries context
|
||||
// properly — see `ClusterNode::stage_signal_local`.
|
||||
reject_unsupported_signal_context(&req).map_err(ClusterAppError)?;
|
||||
// write_signal ships to followers over gRPC (a blocking `runtime.block_on`),
|
||||
// so it must run off the async reactor AND off any thread carrying a runtime
|
||||
// handle — hand it to the runtime-free write pool. A saturated pool yields
|
||||
@ -658,6 +689,9 @@ pub async fn sharded_write_signal(
|
||||
Json(req): Json<SignalRequest>,
|
||||
) -> std::result::Result<StatusCode, ClusterAppError> {
|
||||
require_local_ack(&headers, "/sharded/signals").map_err(ClusterAppError)?;
|
||||
// Same reason as `write_signal`: the scatter-gather write applies
|
||||
// (signal, entity, weight) only and would discard context behind a 204.
|
||||
reject_unsupported_signal_context(&req).map_err(ClusterAppError)?;
|
||||
let shards = state.shard_ids().map_err(ClusterAppError)?;
|
||||
// Offload the blocking single-shard signal write off the reactor (see
|
||||
// [`sharded_create_item`]).
|
||||
|
||||
@ -1,12 +1,9 @@
|
||||
//! Shared request/response DTOs and result-mapping for the data routes.
|
||||
//!
|
||||
//! The standalone router ([`crate::router`]) and the cluster router
|
||||
//! ([`crate::cluster`]) expose the SAME data surface (`/items`,
|
||||
//! `/embeddings`, `/signals`, `/feed`, `/search`). Their request/response
|
||||
//! shapes and the engine-result → JSON mapping were previously duplicated
|
||||
//! verbatim across both modules; any drift between the two (e.g. a new field
|
||||
//! added on one side only) would silently change one mode's API. This module
|
||||
//! is the single source of truth so the two routers cannot diverge.
|
||||
//! The standalone router ([`crate::router`]) and cluster routers
|
||||
//! ([`crate::cluster`]) share these DTOs wherever they expose the same data
|
||||
//! route. The standalone-only exact-rank wire contract lives separately in
|
||||
//! [`crate::exact_rank`] so it cannot leak into a cluster surface.
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
@ -76,10 +73,15 @@ pub struct SignalRequest {
|
||||
/// Signal weight applied to the running decay score.
|
||||
#[schema(example = 1.0)]
|
||||
pub weight: f64,
|
||||
/// Optional originating user context (standalone path only).
|
||||
/// Optional originating user context.
|
||||
///
|
||||
/// Honoured by standalone and by the deployed multi-process cluster. The
|
||||
/// experimental single-process cluster router refuses a request carrying
|
||||
/// it rather than dropping it (`/signals` → 400).
|
||||
#[serde(default)]
|
||||
pub user_id: Option<u64>,
|
||||
/// Optional originating creator context (standalone path only).
|
||||
/// Optional originating creator context. Same support matrix as
|
||||
/// [`Self::user_id`]; drives the `(user, creator)` interaction weight.
|
||||
#[serde(default)]
|
||||
pub creator_id: Option<u64>,
|
||||
}
|
||||
|
||||
191
tidal-server/src/exact_rank.rs
Normal file
191
tidal-server/src/exact_rank.rs
Normal file
@ -0,0 +1,191 @@
|
||||
//! Standalone-only exact-rank HTTP contract.
|
||||
//!
|
||||
//! This module owns the wire translation because timestamps must be decimal
|
||||
//! strings in JSON while the embedded engine uses lossless `u64` nanoseconds.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use axum::{Json, extract::State};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use utoipa::ToSchema;
|
||||
|
||||
use crate::{error::ServerError, router::AppError, state::ServerState};
|
||||
|
||||
/// `POST /rank` body. This route exists only on the standalone server.
|
||||
#[derive(Debug, Serialize, Deserialize, ToSchema)]
|
||||
pub struct ExactRankRequest {
|
||||
/// Frozen profile identity; must be `contest_qualified_hot`.
|
||||
#[schema(example = "contest_qualified_hot")]
|
||||
pub profile: String,
|
||||
/// Frozen formula version; must be `1`.
|
||||
#[schema(example = 1)]
|
||||
pub profile_version: u32,
|
||||
/// Required query clock as decimal Unix-epoch nanoseconds.
|
||||
#[schema(example = "2000000000000000000")]
|
||||
pub as_of_nanos: String,
|
||||
/// Complete candidate set; empty is valid, maximum length is 10,000.
|
||||
pub candidates: Vec<ExactRankCandidate>,
|
||||
}
|
||||
|
||||
/// One caller-supplied candidate in a `POST /rank` request.
|
||||
#[derive(Debug, Serialize, Deserialize, ToSchema)]
|
||||
pub struct ExactRankCandidate {
|
||||
pub entity_id: u64,
|
||||
/// Creation time as decimal Unix-epoch nanoseconds.
|
||||
#[schema(example = "1999913600000000000")]
|
||||
pub created_at_nanos: String,
|
||||
pub signals: ExactRankSignals,
|
||||
}
|
||||
|
||||
/// Raw signals used by exact qualified-hot ranking.
|
||||
#[derive(Debug, Serialize, Deserialize, ToSchema)]
|
||||
pub struct ExactRankSignals {
|
||||
pub hearts: f64,
|
||||
pub comments: f64,
|
||||
/// Delivered exposure, reported nondestructively as a diagnostic only.
|
||||
pub exposure: f64,
|
||||
}
|
||||
|
||||
/// Successful `POST /rank` response.
|
||||
#[derive(Debug, Serialize, ToSchema)]
|
||||
pub struct ExactRankResponse {
|
||||
pub profile: String,
|
||||
pub profile_version: u32,
|
||||
/// Effective query clock as decimal Unix-epoch nanoseconds.
|
||||
#[schema(example = "2000000000000000000")]
|
||||
pub as_of_nanos: String,
|
||||
/// Every input candidate exactly once, best-first.
|
||||
pub items: Vec<ExactRankItem>,
|
||||
}
|
||||
|
||||
/// One item in an exact ranking.
|
||||
#[derive(Debug, Serialize, ToSchema)]
|
||||
pub struct ExactRankItem {
|
||||
pub entity_id: u64,
|
||||
pub rank: usize,
|
||||
pub score: f64,
|
||||
pub components: ExactRankComponents,
|
||||
}
|
||||
|
||||
/// Auditable raw and derived qualified-hot score components.
|
||||
#[derive(Debug, Serialize, ToSchema)]
|
||||
pub struct ExactRankComponents {
|
||||
pub hearts: f64,
|
||||
pub comments: f64,
|
||||
pub exposure: f64,
|
||||
pub engagement: f64,
|
||||
/// Bayesian response diagnostic; not a v1 score input.
|
||||
pub response_rate: f64,
|
||||
/// Multiplicative age factor applied to `ln1p(engagement)`.
|
||||
pub freshness: f64,
|
||||
pub age_hours: f64,
|
||||
}
|
||||
|
||||
impl TryFrom<ExactRankRequest> for tidaldb::query::ExactRankRequest {
|
||||
type Error = ServerError;
|
||||
|
||||
fn try_from(value: ExactRankRequest) -> Result<Self, Self::Error> {
|
||||
if value.candidates.len() > tidaldb::query::MAX_EXACT_RANK_CANDIDATES {
|
||||
return Err(ServerError::BadRequest(format!(
|
||||
"exact ranking accepts at most {} candidates; got {}",
|
||||
tidaldb::query::MAX_EXACT_RANK_CANDIDATES,
|
||||
value.candidates.len()
|
||||
)));
|
||||
}
|
||||
|
||||
let as_of_nanos = parse_decimal_nanos(&value.as_of_nanos, format_args!("as_of_nanos"))?;
|
||||
let candidates = value
|
||||
.candidates
|
||||
.into_iter()
|
||||
.enumerate()
|
||||
.map(|(index, candidate)| {
|
||||
Ok(tidaldb::query::ExactRankCandidate {
|
||||
entity_id: candidate.entity_id,
|
||||
created_at_nanos: parse_decimal_nanos(
|
||||
&candidate.created_at_nanos,
|
||||
format_args!("candidates[{index}].created_at_nanos"),
|
||||
)?,
|
||||
signals: tidaldb::query::ExactRankSignals {
|
||||
hearts: candidate.signals.hearts,
|
||||
comments: candidate.signals.comments,
|
||||
exposure: candidate.signals.exposure,
|
||||
},
|
||||
})
|
||||
})
|
||||
.collect::<Result<Vec<_>, ServerError>>()?;
|
||||
|
||||
Ok(Self {
|
||||
profile: value.profile,
|
||||
profile_version: value.profile_version,
|
||||
as_of_nanos,
|
||||
candidates,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_decimal_nanos(raw: &str, field: std::fmt::Arguments<'_>) -> crate::error::Result<u64> {
|
||||
if raw.is_empty() || !raw.bytes().all(|byte| byte.is_ascii_digit()) {
|
||||
return Err(ServerError::BadRequest(format!(
|
||||
"{field} must contain only decimal digits"
|
||||
)));
|
||||
}
|
||||
raw.parse::<u64>().map_err(|error| {
|
||||
ServerError::BadRequest(format!(
|
||||
"{field} must be an unsigned decimal u64 nanosecond string: {error}"
|
||||
))
|
||||
})
|
||||
}
|
||||
|
||||
impl From<tidaldb::query::ExactRankResponse> for ExactRankResponse {
|
||||
fn from(value: tidaldb::query::ExactRankResponse) -> Self {
|
||||
Self {
|
||||
profile: value.profile,
|
||||
profile_version: value.profile_version,
|
||||
as_of_nanos: value.as_of_nanos.to_string(),
|
||||
items: value
|
||||
.items
|
||||
.into_iter()
|
||||
.map(|item| ExactRankItem {
|
||||
entity_id: item.entity_id,
|
||||
rank: item.rank,
|
||||
score: item.score,
|
||||
components: ExactRankComponents {
|
||||
hearts: item.components.hearts,
|
||||
comments: item.components.comments,
|
||||
exposure: item.components.exposure,
|
||||
engagement: item.components.engagement,
|
||||
response_rate: item.components.response_rate,
|
||||
freshness: item.components.freshness,
|
||||
age_hours: item.components.age_hours,
|
||||
},
|
||||
})
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Rank a complete caller-supplied candidate set without reading database
|
||||
/// candidate, item, ledger, or user state.
|
||||
#[utoipa::path(
|
||||
post,
|
||||
path = "/rank",
|
||||
tag = "data",
|
||||
request_body = ExactRankRequest,
|
||||
responses(
|
||||
(status = 200, description = "Every supplied candidate ranked exactly once", body = ExactRankResponse),
|
||||
(status = 400, description = "Invalid profile, timestamp, candidate, signal, or candidate count"),
|
||||
(status = 401, description = "Missing or invalid API key"),
|
||||
),
|
||||
security(("bearerAuth" = [])),
|
||||
)]
|
||||
pub(crate) async fn rank_exact(
|
||||
State(state): State<Arc<ServerState>>,
|
||||
Json(request): Json<ExactRankRequest>,
|
||||
) -> Result<Json<ExactRankResponse>, AppError> {
|
||||
let request = tidaldb::query::ExactRankRequest::try_from(request).map_err(AppError::from)?;
|
||||
let offload_state = Arc::clone(&state);
|
||||
let response = crate::offload::offload_read(move || offload_state.rank_exact(&request))
|
||||
.await
|
||||
.map_err(AppError::from)?;
|
||||
Ok(Json(response.into()))
|
||||
}
|
||||
@ -1,28 +1,19 @@
|
||||
//! tidal-server — standalone gRPC cluster server for tidaldb.
|
||||
//!
|
||||
//! Wraps the embedded `tidaldb` engine in an axum HTTP surface plus a
|
||||
//! tonic gRPC cluster layer (via `tidal-net`) for multi-shard deployments.
|
||||
//! Carries a Dockerfile + k8s manifests under `tidal/docker/` for standalone
|
||||
//! deployment; embedding consumers use the `tidaldb` library directly and do
|
||||
//! NOT need this server.
|
||||
//!
|
||||
//! Status: M8 Phase 8 and 10 marked PARTIAL; see `tidal/AGENTS.md` § Known
|
||||
//! Gaps for G1/G2/G3.
|
||||
//! Standalone axum HTTP and tonic gRPC server for the embedded `tidaldb` engine.
|
||||
|
||||
/// Public modules exposed for integration testing.
|
||||
///
|
||||
/// The binary (`main.rs`) imports from this lib crate rather than declaring
|
||||
/// the modules directly, so integration tests in `tests/` can access them.
|
||||
/// Public modules used by the binary and integration tests.
|
||||
pub mod cluster;
|
||||
pub mod config;
|
||||
pub mod dto;
|
||||
pub mod error;
|
||||
pub mod exact_rank;
|
||||
pub mod health;
|
||||
/// HTTP surface metrics (requests by route/method/status, per-route latency),
|
||||
/// published through the engine's existing `/metrics` listener.
|
||||
/// HTTP request and per-route latency metrics.
|
||||
///
|
||||
/// Published through the engine's existing `/metrics` listener.
|
||||
pub mod http_metrics;
|
||||
/// Log initialisation: ANSI-free text, or the collector's JSON wire format
|
||||
/// under `JSON_LOGS=1`.
|
||||
/// Log initialisation.
|
||||
///
|
||||
/// Emits ANSI-free text, or collector JSON when `JSON_LOGS=1`.
|
||||
pub mod logging;
|
||||
pub mod offload;
|
||||
pub mod openapi;
|
||||
|
||||
@ -1,16 +1,16 @@
|
||||
//! Machine-readable `OpenAPI` 3.1 specification for the tidalDB HTTP API.
|
||||
//!
|
||||
//! Two [`OpenApi`](utoipa::OpenApi) documents are derived from the
|
||||
//! `#[utoipa::path(...)]` attributes on the handlers and the
|
||||
//! `#[derive(ToSchema)]` on the DTOs:
|
||||
//! Three [`OpenApi`](utoipa::OpenApi) documents are derived from handler paths
|
||||
//! and DTO schemas:
|
||||
//!
|
||||
//! * [`StandaloneApiDoc`] — the data + health surface served by
|
||||
//! [`crate::router::build_router`].
|
||||
//! * [`ClusterApiDoc`] — the same surface PLUS the cluster-management and
|
||||
//! sharded (scatter-gather) routes served by
|
||||
//! [`crate::cluster::build_cluster_router`].
|
||||
//! * [`StandaloneApiDoc`] — common data + health routes and standalone-only
|
||||
//! exact ranking (`POST /rank`).
|
||||
//! * [`ClusterApiDoc`] — common data + health routes plus in-process cluster
|
||||
//! management and sharded routes; deliberately no exact ranking.
|
||||
//! * [`RegionApiDoc`] — the multi-process region-node surface; deliberately no
|
||||
//! exact ranking.
|
||||
//!
|
||||
//! Both are served UNAUTHENTICATED at `GET /openapi.json` (sibling to the
|
||||
//! All three are served UNAUTHENTICATED at `GET /openapi.json` (sibling to the
|
||||
//! `/health/*` probes): the document describes the API contract only — it
|
||||
//! carries no entity data, signals, or secrets — so gating it behind the same
|
||||
//! bearer token a client needs the spec to learn how to send would be
|
||||
@ -33,9 +33,10 @@ pub(crate) async fn serve_standalone() -> Json<utoipa::openapi::OpenApi> {
|
||||
Json(StandaloneApiDoc::openapi())
|
||||
}
|
||||
|
||||
/// `GET /openapi.json` handler for the **cluster** surface — the superset
|
||||
/// document that also describes the `/cluster/*` and `/sharded/*` routes.
|
||||
/// Mounted by [`crate::cluster::build_cluster_router`].
|
||||
/// `GET /openapi.json` handler for the **cluster** surface.
|
||||
///
|
||||
/// Cluster mode adds `/cluster/*` and `/sharded/*` but intentionally omits the
|
||||
/// standalone-only `/rank` endpoint.
|
||||
pub(crate) async fn serve_cluster() -> Json<utoipa::openapi::OpenApi> {
|
||||
Json(ClusterApiDoc::openapi())
|
||||
}
|
||||
@ -79,15 +80,16 @@ impl Modify for SecurityAddon {
|
||||
version = env!("CARGO_PKG_VERSION"),
|
||||
description = "Embeddable, single-node-first database for the personalized \
|
||||
content ranking problem. Write items, embeddings, and signals; \
|
||||
retrieve ranked feeds and run hybrid search. Data routes require \
|
||||
a Bearer token when TIDAL_API_KEY is configured; health probes \
|
||||
and this spec do not.",
|
||||
rank exact caller-owned sets, retrieve feeds, and run hybrid \
|
||||
search. Data routes require a Bearer token when TIDAL_API_KEY \
|
||||
is configured; health probes and this spec do not.",
|
||||
),
|
||||
paths(
|
||||
crate::router::health,
|
||||
crate::router::create_item,
|
||||
crate::router::write_embedding,
|
||||
crate::router::write_signal,
|
||||
crate::exact_rank::rank_exact,
|
||||
crate::router::feed,
|
||||
crate::router::search,
|
||||
crate::router::vector_search,
|
||||
@ -96,6 +98,12 @@ impl Modify for SecurityAddon {
|
||||
crate::dto::ItemRequest,
|
||||
crate::dto::EmbeddingRequest,
|
||||
crate::dto::SignalRequest,
|
||||
crate::exact_rank::ExactRankRequest,
|
||||
crate::exact_rank::ExactRankCandidate,
|
||||
crate::exact_rank::ExactRankSignals,
|
||||
crate::exact_rank::ExactRankResponse,
|
||||
crate::exact_rank::ExactRankItem,
|
||||
crate::exact_rank::ExactRankComponents,
|
||||
crate::dto::FeedResponse,
|
||||
crate::dto::FeedItem,
|
||||
crate::dto::SignalValue,
|
||||
@ -115,13 +123,11 @@ pub struct StandaloneApiDoc;
|
||||
|
||||
/// `OpenAPI` document for the cluster HTTP surface.
|
||||
///
|
||||
/// Superset of [`StandaloneApiDoc`]: the same data + health routes plus the
|
||||
/// cluster-management (`/cluster/*`) and sharded scatter-gather
|
||||
/// (`/sharded/*`) routes. The cluster handlers register the SAME `/items`,
|
||||
/// `/embeddings`, `/signals`, `/feed`, `/search` paths as the standalone ones;
|
||||
/// because this is a distinct document the shared paths are described once here
|
||||
/// from the cluster handlers' attributes (region-aware response text), so there
|
||||
/// is no path-key collision.
|
||||
/// Shares the ordinary data + health routes with [`StandaloneApiDoc`] and adds
|
||||
/// cluster-management (`/cluster/*`) and sharded scatter-gather (`/sharded/*`)
|
||||
/// routes, but omits standalone-only exact ranking. The shared paths are
|
||||
/// described here from cluster handlers with region-aware response text, so
|
||||
/// there is no path-key collision.
|
||||
#[derive(OpenApi)]
|
||||
#[openapi(
|
||||
info(
|
||||
@ -271,12 +277,31 @@ mod tests {
|
||||
"/items",
|
||||
"/embeddings",
|
||||
"/signals",
|
||||
"/rank",
|
||||
"/feed",
|
||||
"/search",
|
||||
"/vector_search",
|
||||
] {
|
||||
assert!(paths.contains_key(p), "standalone doc missing path {p}");
|
||||
}
|
||||
let schemas = &doc
|
||||
.components
|
||||
.as_ref()
|
||||
.expect("components present after schema registration")
|
||||
.schemas;
|
||||
for schema in [
|
||||
"ExactRankRequest",
|
||||
"ExactRankCandidate",
|
||||
"ExactRankSignals",
|
||||
"ExactRankResponse",
|
||||
"ExactRankItem",
|
||||
"ExactRankComponents",
|
||||
] {
|
||||
assert!(
|
||||
schemas.contains_key(schema),
|
||||
"standalone doc missing exact-rank schema {schema}"
|
||||
);
|
||||
}
|
||||
// The cluster-only routes must NOT leak into the standalone document.
|
||||
assert!(
|
||||
!paths.contains_key("/cluster/status"),
|
||||
@ -293,7 +318,7 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// The cluster document is a superset: data + health + cluster + sharded.
|
||||
/// The cluster document adds cluster + sharded routes but omits `/rank`.
|
||||
#[test]
|
||||
fn cluster_doc_lists_cluster_and_sharded_paths() {
|
||||
let doc = ClusterApiDoc::openapi();
|
||||
@ -310,6 +335,10 @@ mod tests {
|
||||
] {
|
||||
assert!(paths.contains_key(p), "cluster doc missing path {p}");
|
||||
}
|
||||
assert!(
|
||||
!paths.contains_key("/rank"),
|
||||
"cluster doc must omit standalone /rank"
|
||||
);
|
||||
}
|
||||
|
||||
/// The multi-process region document enumerates the region node's full
|
||||
@ -343,6 +372,10 @@ mod tests {
|
||||
] {
|
||||
assert!(paths.contains_key(p), "region doc missing path {p}");
|
||||
}
|
||||
assert!(
|
||||
!paths.contains_key("/rank"),
|
||||
"region doc must omit standalone /rank"
|
||||
);
|
||||
let schemes = &doc
|
||||
.components
|
||||
.as_ref()
|
||||
|
||||
@ -56,11 +56,11 @@ pub(crate) const BODY_LIMIT_BYTES: usize = 2 * 1024 * 1024;
|
||||
/// divergence unhealable in production: `POST /cluster/reconcile` failed with
|
||||
/// `503 region unreachable: reconcile peer returned 413 Payload Too Large`,
|
||||
/// which reads like a network fault and sent the operator to TLS and
|
||||
/// NetworkPolicy first.
|
||||
/// `NetworkPolicy` first.
|
||||
///
|
||||
/// This route is internal (`x-tidal-internal` marker), authenticated, and
|
||||
/// driven only by an operator verb, so a large body here is a control-plane
|
||||
/// cost, not an exposed DoS surface.
|
||||
/// cost, not an exposed `DoS` surface.
|
||||
///
|
||||
/// This is a CEILING, not a design: the snapshot grows with the corpus and at
|
||||
/// ~1M entities it will outgrow this too. The durable fix is a chunked
|
||||
@ -138,6 +138,7 @@ pub fn build_router(
|
||||
.route("/items", post(create_item))
|
||||
.route("/embeddings", post(write_embedding))
|
||||
.route("/signals", post(write_signal))
|
||||
.route("/rank", post(crate::exact_rank::rank_exact))
|
||||
.route("/feed", get(feed))
|
||||
.route("/search", get(search))
|
||||
.route("/vector_search", post(vector_search))
|
||||
@ -619,7 +620,13 @@ impl From<TidalErrorWrapper> for AppError {
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct AppError(ServerError);
|
||||
pub(crate) struct AppError(pub(crate) ServerError);
|
||||
|
||||
impl From<ServerError> for AppError {
|
||||
fn from(value: ServerError) -> Self {
|
||||
Self(value)
|
||||
}
|
||||
}
|
||||
|
||||
impl IntoResponse for AppError {
|
||||
fn into_response(self) -> Response {
|
||||
|
||||
@ -8,7 +8,7 @@ use std::{
|
||||
|
||||
use tidaldb::{
|
||||
TidalDb,
|
||||
query::{retrieve::Retrieve, search::Search},
|
||||
query::{ExactRankRequest, ExactRankResponse, retrieve::Retrieve, search::Search},
|
||||
schema::EntityId,
|
||||
};
|
||||
|
||||
@ -125,6 +125,19 @@ impl ServerState {
|
||||
}
|
||||
}
|
||||
|
||||
/// Execute the standalone exact-rank formula over caller-owned candidates.
|
||||
///
|
||||
/// This delegates to the engine's pure exact-rank API and reads no stored
|
||||
/// item, ledger, candidate, or user state.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns [`ServerError`] when the exact-rank request violates its pinned
|
||||
/// profile, clock, candidate, numeric, or result-size contract.
|
||||
pub fn rank_exact(&self, request: &ExactRankRequest) -> Result<ExactRankResponse> {
|
||||
self.db.rank_exact(request).map_err(ServerError::from)
|
||||
}
|
||||
|
||||
/// Retrieve and rank items for the (standalone) region.
|
||||
///
|
||||
/// # Errors
|
||||
|
||||
167
tidal-server/tests/exact_rank.rs
Normal file
167
tidal-server/tests/exact_rank.rs
Normal file
@ -0,0 +1,167 @@
|
||||
#![allow(clippy::unwrap_used)]
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use axum::{
|
||||
body::Body,
|
||||
http::{Method, Request, StatusCode},
|
||||
};
|
||||
use tidal_server::{router::build_router, state::ServerState};
|
||||
use tidaldb::TidalDb;
|
||||
use tower::ServiceExt;
|
||||
|
||||
const AS_OF: &str = "2000000000000000000";
|
||||
|
||||
fn make_app() -> axum::Router {
|
||||
let (schema, profiles) = tidal_server::config::load_schema(None).unwrap();
|
||||
let db = TidalDb::builder()
|
||||
.ephemeral()
|
||||
.with_schema(schema)
|
||||
.with_profiles(profiles)
|
||||
.open()
|
||||
.unwrap();
|
||||
build_router(
|
||||
Arc::new(ServerState::new(db)),
|
||||
Arc::new(tidal_server::cluster::security::ClusterCreds::unauthenticated()),
|
||||
)
|
||||
}
|
||||
|
||||
async fn post_rank(body: serde_json::Value) -> (StatusCode, serde_json::Value) {
|
||||
let response = make_app()
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.method(Method::POST)
|
||||
.uri("/rank")
|
||||
.header("Content-Type", "application/json")
|
||||
.body(Body::from(serde_json::to_vec(&body).unwrap()))
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let status = response.status();
|
||||
let bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
|
||||
.await
|
||||
.unwrap();
|
||||
let body = serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null);
|
||||
(status, body)
|
||||
}
|
||||
|
||||
fn rank_body(candidates: &serde_json::Value) -> serde_json::Value {
|
||||
serde_json::json!({
|
||||
"profile": "contest_qualified_hot",
|
||||
"profile_version": 1,
|
||||
"as_of_nanos": AS_OF,
|
||||
"candidates": candidates,
|
||||
})
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rank_wire_contract_is_lossless_complete_and_deterministic() {
|
||||
let (status, body) = post_rank(rank_body(&serde_json::json!([
|
||||
{
|
||||
"entity_id": 1,
|
||||
"created_at_nanos": "1999913600000000000",
|
||||
"signals": { "hearts": 8.0, "comments": 4.0, "exposure": 100.0 }
|
||||
},
|
||||
{
|
||||
"entity_id": 9,
|
||||
"created_at_nanos": "1999913600000000000",
|
||||
"signals": { "hearts": 8.0, "comments": 4.0, "exposure": 100_000.0 }
|
||||
}
|
||||
])))
|
||||
.await;
|
||||
|
||||
assert_eq!(status, StatusCode::OK, "body: {body}");
|
||||
assert_eq!(body["profile"], "contest_qualified_hot");
|
||||
assert_eq!(body["profile_version"], 1);
|
||||
assert_eq!(body["as_of_nanos"], AS_OF);
|
||||
let items = body["items"].as_array().unwrap();
|
||||
assert_eq!(items.len(), 2);
|
||||
assert_eq!(items[0]["entity_id"], 9);
|
||||
assert_eq!(items[0]["rank"], 1);
|
||||
assert_eq!(items[1]["entity_id"], 1);
|
||||
assert_eq!(items[1]["rank"], 2);
|
||||
assert_eq!(items[0]["score"], items[1]["score"]);
|
||||
assert_ne!(
|
||||
items[0]["components"]["response_rate"],
|
||||
items[1]["components"]["response_rate"]
|
||||
);
|
||||
for item in items {
|
||||
for field in [
|
||||
"hearts",
|
||||
"comments",
|
||||
"exposure",
|
||||
"engagement",
|
||||
"response_rate",
|
||||
"freshness",
|
||||
"age_hours",
|
||||
] {
|
||||
assert!(
|
||||
item["components"].get(field).is_some(),
|
||||
"missing {field}: {item}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rank_requires_a_pinned_clock() {
|
||||
let mut request = rank_body(&serde_json::json!([]));
|
||||
request.as_object_mut().unwrap().remove("as_of_nanos");
|
||||
assert!(post_rank(request).await.0.is_client_error());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rank_rejects_unpinned_profile_invalid_candidates_and_numeric_timestamps() {
|
||||
let mut wrong_version = rank_body(&serde_json::json!([]));
|
||||
wrong_version["profile_version"] = serde_json::json!(2);
|
||||
assert_eq!(post_rank(wrong_version).await.0, StatusCode::BAD_REQUEST);
|
||||
|
||||
let duplicate = serde_json::json!([
|
||||
{ "entity_id": 1, "created_at_nanos": AS_OF, "signals": { "hearts": 1.0, "comments": 0.0, "exposure": 1.0 } },
|
||||
{ "entity_id": 1, "created_at_nanos": AS_OF, "signals": { "hearts": 2.0, "comments": 0.0, "exposure": 2.0 } }
|
||||
]);
|
||||
assert_eq!(
|
||||
post_rank(rank_body(&duplicate)).await.0,
|
||||
StatusCode::BAD_REQUEST
|
||||
);
|
||||
|
||||
let future = serde_json::json!([
|
||||
{ "entity_id": 1, "created_at_nanos": "2000000000000000001", "signals": { "hearts": 1.0, "comments": 0.0, "exposure": 1.0 } }
|
||||
]);
|
||||
assert_eq!(
|
||||
post_rank(rank_body(&future)).await.0,
|
||||
StatusCode::BAD_REQUEST
|
||||
);
|
||||
|
||||
let negative = serde_json::json!([
|
||||
{ "entity_id": 1, "created_at_nanos": AS_OF, "signals": { "hearts": -1.0, "comments": 0.0, "exposure": 1.0 } }
|
||||
]);
|
||||
assert_eq!(
|
||||
post_rank(rank_body(&negative)).await.0,
|
||||
StatusCode::BAD_REQUEST
|
||||
);
|
||||
|
||||
let mut numeric = rank_body(&serde_json::json!([]));
|
||||
numeric["as_of_nanos"] = serde_json::json!(2_000_000_000_000_000_000u64);
|
||||
assert!(post_rank(numeric).await.0.is_client_error());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rank_rejects_an_oversized_set_before_per_candidate_conversion() {
|
||||
let candidates = (0..10_001)
|
||||
.map(|entity_id| {
|
||||
serde_json::json!({
|
||||
"entity_id": entity_id,
|
||||
"created_at_nanos": "not-a-timestamp",
|
||||
"signals": { "hearts": 0.0, "comments": 0.0, "exposure": 0.0 }
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
let (status, body) = post_rank(rank_body(&serde_json::Value::Array(candidates))).await;
|
||||
|
||||
assert_eq!(status, StatusCode::BAD_REQUEST);
|
||||
let text = body.to_string();
|
||||
assert!(text.contains("at most 10000 candidates"), "{body}");
|
||||
assert!(!text.contains("created_at_nanos"), "{body}");
|
||||
}
|
||||
@ -186,10 +186,10 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
|
||||
// ── 5. Retrieve ranked results ──────────────────────────────────────
|
||||
|
||||
// The `hot` builtin profile ranks by cumulative view count with age decay
|
||||
// (Reddit/HN-style). Items with more views score higher; the age factor
|
||||
// penalises older content. All items are treated as 24 hours old here
|
||||
// since metadata-based age lookup is wired in M3+.
|
||||
// The `hot` builtin profile ranks cumulative view count with age decay.
|
||||
// Items with more views score higher; the age factor penalises older
|
||||
// content. Age comes from each item's `created_at` metadata, with the
|
||||
// documented 24-hour default used only when that metadata is absent.
|
||||
let query = tidaldb::query::retrieve::Retrieve::builder()
|
||||
.profile("hot")
|
||||
.limit(10)
|
||||
|
||||
@ -22,6 +22,7 @@ mod open;
|
||||
pub mod paths;
|
||||
mod purge;
|
||||
mod query_ops;
|
||||
mod rank;
|
||||
mod relationships;
|
||||
mod rematerialization;
|
||||
mod remove_scope;
|
||||
|
||||
33
tidal/src/db/rank.rs
Normal file
33
tidal/src/db/rank.rs
Normal file
@ -0,0 +1,33 @@
|
||||
//! Database entry point for exact ranking.
|
||||
|
||||
use super::TidalDb;
|
||||
use crate::query::rank::{ExactRankRequest, ExactRankResponse, execute};
|
||||
|
||||
impl TidalDb {
|
||||
/// Execute the frozen `contest_qualified_hot` v1 formula over a complete,
|
||||
/// caller-supplied candidate set.
|
||||
///
|
||||
/// This method reads no database state: no configured ranking profiles,
|
||||
/// candidates, items, signal ledger, or users. The explicit profile name and
|
||||
/// version in the request pin formula semantics against operator config
|
||||
/// changes.
|
||||
/// The required `as_of_nanos` is evaluated once by the caller and remains
|
||||
/// part of the request identity.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns [`crate::TidalError::InvalidInput`] for an unsupported profile
|
||||
/// identity, invalid timestamp/candidate/signal, duplicate entity ID,
|
||||
/// arithmetic overflow, or a candidate set larger than 10,000.
|
||||
#[tracing::instrument(
|
||||
skip_all,
|
||||
fields(
|
||||
profile = %request.profile,
|
||||
profile_version = request.profile_version,
|
||||
candidates = request.candidates.len()
|
||||
)
|
||||
)]
|
||||
pub fn rank_exact(&self, request: &ExactRankRequest) -> crate::Result<ExactRankResponse> {
|
||||
execute(request)
|
||||
}
|
||||
}
|
||||
@ -18,12 +18,32 @@ use crate::{
|
||||
signals::StagedLedgerApply,
|
||||
};
|
||||
|
||||
/// The originating user/creator context of a signal write.
|
||||
///
|
||||
/// Carried on a [`StagedSignal`] so the two-phase path applies the SAME
|
||||
/// side effects as [`TidalDb::signal_with_context`] — see
|
||||
/// [`TidalDb::apply_signal_context`]. Owns its signal-type name because the
|
||||
/// staged write outlives the borrow of the request that produced it.
|
||||
#[derive(Debug)]
|
||||
struct SignalContext {
|
||||
signal_type: String,
|
||||
entity_id: EntityId,
|
||||
weight: f64,
|
||||
timestamp: Timestamp,
|
||||
for_user: Option<u64>,
|
||||
creator_id: Option<u64>,
|
||||
}
|
||||
|
||||
/// A staged signal write on a [`TidalDb`]: admission-checked and WAL-submitted,
|
||||
/// durability and in-memory fold pending. Created by
|
||||
/// [`TidalDb::signal_staged`]; completed by [`wait`](Self::wait).
|
||||
/// [`TidalDb::signal_staged`] or
|
||||
/// [`TidalDb::signal_with_context_staged`]; completed by [`wait`](Self::wait).
|
||||
#[derive(Debug)]
|
||||
pub struct StagedSignal {
|
||||
staged: StagedLedgerApply,
|
||||
/// `Some` only for a `signal_with_context_staged` write. `None` leaves
|
||||
/// `wait` byte-for-byte equivalent to the pre-context behaviour.
|
||||
context: Option<SignalContext>,
|
||||
#[cfg(feature = "metrics")]
|
||||
write_start: std::time::Instant,
|
||||
}
|
||||
@ -33,6 +53,12 @@ impl StagedSignal {
|
||||
/// in-memory aggregate (identical end state to a completed
|
||||
/// [`TidalDb::signal`] call, including the write-latency metrics).
|
||||
///
|
||||
/// For a write staged with originating context, the user/creator side
|
||||
/// effects (hard negatives, seen, interaction weight, preference vector,
|
||||
/// cohort and community forwarding) are applied here, AFTER durability —
|
||||
/// the same order [`TidalDb::signal_with_context`] uses, so a crash can
|
||||
/// never leave a side effect whose base signal was never logged.
|
||||
///
|
||||
/// Returns the event's assigned WAL seqno — the replicated-stream
|
||||
/// position quorum acks gate on (m11p3). `0` = suppressed by the dedup
|
||||
/// window (an identical record is already durable; its quorum status is
|
||||
@ -47,6 +73,19 @@ impl StagedSignal {
|
||||
pub fn wait(self, db: &TidalDb) -> crate::Result<u64> {
|
||||
let result = db.ledger()?.complete_staged(self.staged);
|
||||
|
||||
if result.is_ok()
|
||||
&& let Some(ctx) = self.context
|
||||
{
|
||||
db.apply_signal_context(
|
||||
&ctx.signal_type,
|
||||
ctx.entity_id,
|
||||
ctx.weight,
|
||||
ctx.timestamp,
|
||||
ctx.for_user,
|
||||
ctx.creator_id,
|
||||
);
|
||||
}
|
||||
|
||||
#[cfg(feature = "metrics")]
|
||||
if result.is_ok() {
|
||||
use std::sync::atomic::Ordering;
|
||||
@ -277,7 +316,74 @@ impl TidalDb {
|
||||
weight: f64,
|
||||
timestamp: Timestamp,
|
||||
) -> crate::Result<StagedSignal> {
|
||||
self.require_writeable("signal_staged")?;
|
||||
self.stage_signal_inner(
|
||||
"signal_staged",
|
||||
signal_type,
|
||||
entity_id,
|
||||
weight,
|
||||
timestamp,
|
||||
None,
|
||||
)
|
||||
}
|
||||
|
||||
/// Two-phase counterpart of
|
||||
/// [`signal_with_context`](Self::signal_with_context).
|
||||
///
|
||||
/// Staging validates and submits the base signal; [`StagedSignal::wait`]
|
||||
/// makes it durable and THEN applies the user/creator side effects. This is
|
||||
/// the write a replicated cluster leader needs: `signal_with_context` would
|
||||
/// serialise every context-carrying write on its own fsync, and
|
||||
/// `signal_staged` silently discards the context.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Same admission errors as [`signal_staged`](Self::signal_staged), plus
|
||||
/// `InvalidInput` if `for_user` is set and `entity_id` exceeds the `u32`
|
||||
/// item-universe limit (see `signal_with_context` for why that aliasing is
|
||||
/// rejected up front rather than truncated).
|
||||
pub fn signal_with_context_staged(
|
||||
&self,
|
||||
signal_type: &str,
|
||||
entity_id: EntityId,
|
||||
weight: f64,
|
||||
timestamp: Timestamp,
|
||||
for_user: Option<u64>,
|
||||
creator_id: Option<u64>,
|
||||
) -> crate::Result<StagedSignal> {
|
||||
Self::validate_context_entity(entity_id, for_user)?;
|
||||
// No context at all ⇒ no side effects to apply; take the plain path so
|
||||
// `wait` does not carry a pointless allocation per write.
|
||||
let context = (for_user.is_some() || creator_id.is_some()).then(|| SignalContext {
|
||||
signal_type: signal_type.to_owned(),
|
||||
entity_id,
|
||||
weight,
|
||||
timestamp,
|
||||
for_user,
|
||||
creator_id,
|
||||
});
|
||||
self.stage_signal_inner(
|
||||
"signal_with_context_staged",
|
||||
signal_type,
|
||||
entity_id,
|
||||
weight,
|
||||
timestamp,
|
||||
context,
|
||||
)
|
||||
}
|
||||
|
||||
/// Shared staging body for [`signal_staged`](Self::signal_staged) and
|
||||
/// [`signal_with_context_staged`](Self::signal_with_context_staged): one
|
||||
/// admission sequence, one `StagedSignal` construction.
|
||||
fn stage_signal_inner(
|
||||
&self,
|
||||
op: &'static str,
|
||||
signal_type: &str,
|
||||
entity_id: EntityId,
|
||||
weight: f64,
|
||||
timestamp: Timestamp,
|
||||
context: Option<SignalContext>,
|
||||
) -> crate::Result<StagedSignal> {
|
||||
self.require_writeable(op)?;
|
||||
Self::validate_signal_weight(weight)?;
|
||||
self.check_write_backpressure()?;
|
||||
|
||||
@ -290,6 +396,7 @@ impl TidalDb {
|
||||
|
||||
Ok(StagedSignal {
|
||||
staged,
|
||||
context,
|
||||
#[cfg(feature = "metrics")]
|
||||
write_start,
|
||||
})
|
||||
@ -648,16 +755,39 @@ impl TidalDb {
|
||||
creator_id: Option<u64>,
|
||||
) -> crate::Result<()> {
|
||||
self.require_writeable("signal_with_context")?;
|
||||
Self::validate_context_entity(entity_id, for_user)?;
|
||||
|
||||
// When a `for_user` identity is present this call narrows the item id to
|
||||
// its u32 slot and writes it into DURABLE Tag::HardNeg / Tag::UserState
|
||||
// rows. A bare `as u32` truncation would silently alias two items whose
|
||||
// ids share their low 32 bits — a permanent cross-id collision in the
|
||||
// hard-negative / seen / saved / liked correctness primitives that
|
||||
// survives restart. Reject an over-range id up front (mirroring
|
||||
// `write_item_with_metadata`) so a colliding durable row never lands on
|
||||
// disk. Done before the base `signal()` so a rejected write leaves no
|
||||
// trace at all.
|
||||
// Record the base signal.
|
||||
self.signal(signal_type, entity_id, weight, timestamp)?;
|
||||
|
||||
self.apply_signal_context(
|
||||
signal_type,
|
||||
entity_id,
|
||||
weight,
|
||||
timestamp,
|
||||
for_user,
|
||||
creator_id,
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Reject an item id that cannot round-trip through the `u32` item slot.
|
||||
///
|
||||
/// When a `for_user` identity is present the write narrows the item id to
|
||||
/// its u32 slot and stores it in DURABLE `Tag::HardNeg` / `Tag::UserState`
|
||||
/// rows. A bare `as u32` truncation would silently alias two items whose
|
||||
/// ids share their low 32 bits — a permanent cross-id collision in the
|
||||
/// hard-negative / seen / saved / liked correctness primitives that
|
||||
/// survives restart. Rejecting up front (mirroring
|
||||
/// `write_item_with_metadata`) keeps a colliding durable row off disk, and
|
||||
/// doing it BEFORE the base signal leaves a rejected write with no trace at
|
||||
/// all.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// `TidalError::InvalidInput` if `for_user` is set and `entity_id` exceeds
|
||||
/// `u32::MAX`.
|
||||
fn validate_context_entity(entity_id: EntityId, for_user: Option<u64>) -> crate::Result<()> {
|
||||
if for_user.is_some() && entity_id.as_u64() > u64::from(u32::MAX) {
|
||||
let raw = entity_id.as_u64();
|
||||
return Err(TidalError::invalid_input(format!(
|
||||
@ -666,10 +796,30 @@ impl TidalDb {
|
||||
u32::MAX
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Record the base signal.
|
||||
self.signal(signal_type, entity_id, weight, timestamp)?;
|
||||
|
||||
/// Apply the user/creator side effects of a signal whose base write is
|
||||
/// already durable.
|
||||
///
|
||||
/// Shared by [`signal_with_context`](Self::signal_with_context) and the
|
||||
/// two-phase [`StagedSignal::wait`] so the replicated cluster path and the
|
||||
/// standalone path cannot diverge — the divergence this replaced silently
|
||||
/// dropped `user_id`/`creator_id` on every clustered write while still
|
||||
/// answering 204.
|
||||
///
|
||||
/// Infallible by construction: every step is either in-memory or a
|
||||
/// best-effort durable write that logs its own failure. The base signal has
|
||||
/// already succeeded, so a side-effect failure must not retract it.
|
||||
fn apply_signal_context(
|
||||
&self,
|
||||
signal_type: &str,
|
||||
entity_id: EntityId,
|
||||
weight: f64,
|
||||
timestamp: Timestamp,
|
||||
for_user: Option<u64>,
|
||||
creator_id: Option<u64>,
|
||||
) {
|
||||
// pg1: record user's most recent signal timestamp for staleness/feedback-loop tracking.
|
||||
#[cfg(feature = "metrics")]
|
||||
if let Some(user_id) = for_user {
|
||||
@ -757,8 +907,6 @@ impl TidalDb {
|
||||
// 8. Community forwarding (M9): forward to opted-in community aggregates.
|
||||
self.try_community_forwarding(signal_type, entity_id, weight, timestamp, user_id);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Persist the current `(user, creator)` interaction weight as a durable
|
||||
|
||||
@ -10,6 +10,7 @@
|
||||
|
||||
pub mod executor;
|
||||
pub mod fusion;
|
||||
pub mod rank;
|
||||
pub mod retrieve;
|
||||
pub mod search;
|
||||
pub mod stats;
|
||||
@ -19,6 +20,11 @@ pub use executor::RetrieveExecutor;
|
||||
pub use fusion::{
|
||||
HybridFusion, RetrievalMode, ann_to_ranked, normalize_fused_scores, route_results, rrf_term,
|
||||
};
|
||||
pub use rank::{
|
||||
EXACT_RANK_PROFILE, EXACT_RANK_PROFILE_VERSION, ExactRankCandidate, ExactRankComponents,
|
||||
ExactRankItem, ExactRankRequest, ExactRankResponse, ExactRankSignals,
|
||||
MAX_EXACT_RANK_CANDIDATES,
|
||||
};
|
||||
pub use retrieve::{
|
||||
Cursor, ProfileRef, QueryError, Results, Retrieve, RetrieveBuilder, RetrieveResult, Signal,
|
||||
};
|
||||
|
||||
228
tidal/src/query/rank/mod.rs
Normal file
228
tidal/src/query/rank/mod.rs
Normal file
@ -0,0 +1,228 @@
|
||||
//! Exact ranking over a caller-supplied candidate set.
|
||||
//!
|
||||
//! This path is intentionally independent of RETRIEVE candidate generation,
|
||||
//! ranking-profile configuration, and every database-backed item, signal,
|
||||
//! ledger, and user store. The caller supplies every value used by the frozen
|
||||
//! `contest_qualified_hot` v1 formula.
|
||||
|
||||
use std::collections::HashSet;
|
||||
|
||||
use crate::schema::TidalError;
|
||||
|
||||
/// Only profile accepted by [`crate::TidalDb::rank_exact`].
|
||||
pub const EXACT_RANK_PROFILE: &str = "contest_qualified_hot";
|
||||
/// Only profile version accepted by [`crate::TidalDb::rank_exact`].
|
||||
pub const EXACT_RANK_PROFILE_VERSION: u32 = 1;
|
||||
/// Maximum number of candidates accepted by one exact-rank request.
|
||||
pub const MAX_EXACT_RANK_CANDIDATES: usize = 10_000;
|
||||
|
||||
const GRAVITY: f64 = 0.25;
|
||||
const AGE_OFFSET_HOURS: f64 = 48.0;
|
||||
const COMMENT_WEIGHT: f64 = 0.25;
|
||||
const PRIOR_EXPOSURE: f64 = 20.0;
|
||||
const PRIOR_RESPONSE_RATE: f64 = 0.10;
|
||||
const NANOS_PER_HOUR: f64 = 3_600_000_000_000.0;
|
||||
|
||||
/// Complete input to [`crate::TidalDb::rank_exact`].
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct ExactRankRequest {
|
||||
/// Must equal [`EXACT_RANK_PROFILE`].
|
||||
pub profile: String,
|
||||
/// Must equal [`EXACT_RANK_PROFILE_VERSION`].
|
||||
pub profile_version: u32,
|
||||
/// Required query clock in Unix-epoch nanoseconds.
|
||||
pub as_of_nanos: u64,
|
||||
/// Complete candidate set. No database candidates are added or removed.
|
||||
pub candidates: Vec<ExactRankCandidate>,
|
||||
}
|
||||
|
||||
/// One caller-supplied exact-rank candidate.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct ExactRankCandidate {
|
||||
pub entity_id: u64,
|
||||
pub created_at_nanos: u64,
|
||||
pub signals: ExactRankSignals,
|
||||
}
|
||||
|
||||
/// Raw signals consumed by the qualified-hot formula.
|
||||
#[derive(Debug, Clone, Copy, PartialEq)]
|
||||
pub struct ExactRankSignals {
|
||||
pub hearts: f64,
|
||||
pub comments: f64,
|
||||
pub exposure: f64,
|
||||
}
|
||||
|
||||
/// Complete exact-rank result.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct ExactRankResponse {
|
||||
pub profile: String,
|
||||
pub profile_version: u32,
|
||||
pub as_of_nanos: u64,
|
||||
pub items: Vec<ExactRankItem>,
|
||||
}
|
||||
|
||||
/// One scored candidate, ordered best-first.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct ExactRankItem {
|
||||
pub entity_id: u64,
|
||||
pub rank: usize,
|
||||
pub score: f64,
|
||||
pub components: ExactRankComponents,
|
||||
}
|
||||
|
||||
/// Auditable inputs and intermediate values for an exact score.
|
||||
#[derive(Debug, Clone, Copy, PartialEq)]
|
||||
pub struct ExactRankComponents {
|
||||
pub hearts: f64,
|
||||
pub comments: f64,
|
||||
pub exposure: f64,
|
||||
pub engagement: f64,
|
||||
/// Bayesian response diagnostic; not a v1 score input.
|
||||
pub response_rate: f64,
|
||||
/// Multiplicative age factor applied to `ln1p(engagement)`.
|
||||
pub freshness: f64,
|
||||
pub age_hours: f64,
|
||||
}
|
||||
|
||||
/// Execute the pure exact-rank calculation at the request's pinned clock.
|
||||
pub(crate) fn execute(request: &ExactRankRequest) -> crate::Result<ExactRankResponse> {
|
||||
validate_profile_identity(request)?;
|
||||
validate_candidates(&request.candidates, request.as_of_nanos)?;
|
||||
|
||||
let mut items = Vec::with_capacity(request.candidates.len());
|
||||
for candidate in &request.candidates {
|
||||
items.push(score_candidate(candidate, request.as_of_nanos)?);
|
||||
}
|
||||
|
||||
items.sort_unstable_by(|left, right| {
|
||||
right
|
||||
.score
|
||||
.total_cmp(&left.score)
|
||||
.then_with(|| right.entity_id.cmp(&left.entity_id))
|
||||
});
|
||||
for (index, item) in items.iter_mut().enumerate() {
|
||||
item.rank = index + 1;
|
||||
}
|
||||
|
||||
Ok(ExactRankResponse {
|
||||
profile: request.profile.clone(),
|
||||
profile_version: request.profile_version,
|
||||
as_of_nanos: request.as_of_nanos,
|
||||
items,
|
||||
})
|
||||
}
|
||||
|
||||
fn validate_profile_identity(request: &ExactRankRequest) -> crate::Result<()> {
|
||||
if request.profile != EXACT_RANK_PROFILE {
|
||||
return Err(TidalError::invalid_input(format!(
|
||||
"exact ranking profile must be '{EXACT_RANK_PROFILE}'; got '{}'",
|
||||
request.profile
|
||||
)));
|
||||
}
|
||||
if request.profile_version != EXACT_RANK_PROFILE_VERSION {
|
||||
return Err(TidalError::invalid_input(format!(
|
||||
"exact ranking profile '{EXACT_RANK_PROFILE}' requires version \
|
||||
{EXACT_RANK_PROFILE_VERSION}; got {}",
|
||||
request.profile_version
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_candidates(candidates: &[ExactRankCandidate], as_of_nanos: u64) -> crate::Result<()> {
|
||||
if candidates.len() > MAX_EXACT_RANK_CANDIDATES {
|
||||
return Err(TidalError::invalid_input(format!(
|
||||
"exact ranking accepts at most {MAX_EXACT_RANK_CANDIDATES} candidates; got {}",
|
||||
candidates.len()
|
||||
)));
|
||||
}
|
||||
|
||||
let mut entity_ids = HashSet::with_capacity(candidates.len());
|
||||
for (index, candidate) in candidates.iter().enumerate() {
|
||||
if !entity_ids.insert(candidate.entity_id) {
|
||||
return Err(TidalError::invalid_input(format!(
|
||||
"candidate {index} duplicates entity_id {}",
|
||||
candidate.entity_id
|
||||
)));
|
||||
}
|
||||
if candidate.created_at_nanos > as_of_nanos {
|
||||
return Err(TidalError::invalid_input(format!(
|
||||
"candidate {} created_at_nanos {} is after as_of_nanos {as_of_nanos}",
|
||||
candidate.entity_id, candidate.created_at_nanos
|
||||
)));
|
||||
}
|
||||
for (name, value) in [
|
||||
("hearts", candidate.signals.hearts),
|
||||
("comments", candidate.signals.comments),
|
||||
("exposure", candidate.signals.exposure),
|
||||
] {
|
||||
if !value.is_finite() || value < 0.0 {
|
||||
return Err(TidalError::invalid_input(format!(
|
||||
"candidate {} signal '{name}' must be finite and non-negative",
|
||||
candidate.entity_id
|
||||
)));
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Keep the published arithmetic as ordinary multiply/add operations. On the
|
||||
// baseline x86-64 build, forcing `mul_add` can lower to an out-of-line libm FMA
|
||||
// call for every candidate while changing the contract's rounding semantics.
|
||||
#[allow(clippy::suboptimal_flops)]
|
||||
fn score_candidate(
|
||||
candidate: &ExactRankCandidate,
|
||||
as_of_nanos: u64,
|
||||
) -> crate::Result<ExactRankItem> {
|
||||
#[allow(clippy::cast_precision_loss)]
|
||||
let age_hours = as_of_nanos.saturating_sub(candidate.created_at_nanos) as f64 / NANOS_PER_HOUR;
|
||||
let engagement = candidate.signals.hearts + candidate.signals.comments * COMMENT_WEIGHT;
|
||||
let opportunities = candidate.signals.exposure.max(engagement);
|
||||
let response_rate =
|
||||
(engagement + PRIOR_EXPOSURE * PRIOR_RESPONSE_RATE) / (opportunities + PRIOR_EXPOSURE);
|
||||
let freshness = (AGE_OFFSET_HOURS / (age_hours + AGE_OFFSET_HOURS)).powf(GRAVITY);
|
||||
|
||||
// Exposure intentionally affects only the response-rate diagnostic in v1.
|
||||
// Ranking on it rewards candidates whose delivery system withheld
|
||||
// impressions and creates severe mid-band inversions.
|
||||
let score = if engagement == 0.0 {
|
||||
0.0
|
||||
} else {
|
||||
engagement.ln_1p() * freshness
|
||||
};
|
||||
if [
|
||||
age_hours,
|
||||
engagement,
|
||||
opportunities,
|
||||
response_rate,
|
||||
freshness,
|
||||
score,
|
||||
]
|
||||
.iter()
|
||||
.any(|value| !value.is_finite())
|
||||
{
|
||||
return Err(TidalError::invalid_input(format!(
|
||||
"candidate {} produces a non-finite qualified-hot component",
|
||||
candidate.entity_id
|
||||
)));
|
||||
}
|
||||
|
||||
Ok(ExactRankItem {
|
||||
entity_id: candidate.entity_id,
|
||||
rank: 0,
|
||||
score,
|
||||
components: ExactRankComponents {
|
||||
hearts: candidate.signals.hearts,
|
||||
comments: candidate.signals.comments,
|
||||
exposure: candidate.signals.exposure,
|
||||
engagement,
|
||||
response_rate,
|
||||
freshness,
|
||||
age_hours,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
225
tidal/src/query/rank/tests.rs
Normal file
225
tidal/src/query/rank/tests.rs
Normal file
@ -0,0 +1,225 @@
|
||||
#![allow(
|
||||
clippy::cast_possible_truncation,
|
||||
clippy::float_cmp,
|
||||
clippy::unwrap_used
|
||||
)]
|
||||
|
||||
use std::collections::BTreeSet;
|
||||
|
||||
use super::*;
|
||||
|
||||
const AS_OF: u64 = 2_000_000_000_000_000_000;
|
||||
const HOUR: u64 = 3_600_000_000_000;
|
||||
|
||||
fn candidate(entity_id: u64, age_hours: u64, hearts: f64, exposure: f64) -> ExactRankCandidate {
|
||||
ExactRankCandidate {
|
||||
entity_id,
|
||||
created_at_nanos: AS_OF - age_hours * HOUR,
|
||||
signals: ExactRankSignals {
|
||||
hearts,
|
||||
comments: 0.0,
|
||||
exposure,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
fn request(candidates: Vec<ExactRankCandidate>) -> ExactRankRequest {
|
||||
ExactRankRequest {
|
||||
profile: EXACT_RANK_PROFILE.into(),
|
||||
profile_version: EXACT_RANK_PROFILE_VERSION,
|
||||
as_of_nanos: AS_OF,
|
||||
candidates,
|
||||
}
|
||||
}
|
||||
|
||||
fn run(candidates: Vec<ExactRankCandidate>) -> crate::Result<ExactRankResponse> {
|
||||
execute(&request(candidates))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn deterministic_ties_use_descending_entity_id() {
|
||||
let result = run(vec![
|
||||
candidate(1, 12, 5.0, 10.0),
|
||||
candidate(3, 12, 5.0, 10.0),
|
||||
candidate(2, 12, 5.0, 10.0),
|
||||
])
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
result
|
||||
.items
|
||||
.iter()
|
||||
.map(|item| item.entity_id)
|
||||
.collect::<Vec<_>>(),
|
||||
vec![3, 2, 1]
|
||||
);
|
||||
assert_eq!(
|
||||
result
|
||||
.items
|
||||
.iter()
|
||||
.map(|item| item.rank)
|
||||
.collect::<Vec<_>>(),
|
||||
vec![1, 2, 3]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn zero_support_scores_zero_regardless_of_freshness() {
|
||||
let result = run(vec![
|
||||
candidate(1, 1_000, 0.0, 0.0),
|
||||
candidate(2, 0, 0.0, 5_000.0),
|
||||
])
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result.items[0].entity_id, 2);
|
||||
assert_eq!(result.items[0].score, 0.0);
|
||||
assert_eq!(result.items[1].score, 0.0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn exposure_changes_response_diagnostic_but_not_score() {
|
||||
let result = run(vec![
|
||||
candidate(1, 24, 8.0, 8.0),
|
||||
candidate(2, 24, 8.0, 80_000.0),
|
||||
])
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result.items[0].score, result.items[1].score);
|
||||
assert!(
|
||||
(result.items[0].components.response_rate - result.items[1].components.response_rate).abs()
|
||||
> f64::EPSILON
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn supported_high_volume_entry_beats_small_perfect_entry() {
|
||||
let result = run(vec![
|
||||
candidate(1, 24, 2.0, 3.0),
|
||||
candidate(2, 24, 1_200.0, 3_000.0),
|
||||
])
|
||||
.unwrap();
|
||||
assert_eq!(result.items[0].entity_id, 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mid_band_support_wins_at_equal_age_and_after_thirty_days() {
|
||||
let equal_age = run(vec![
|
||||
candidate(1, 0, 3.0, 0.0),
|
||||
candidate(2, 0, 40.0, 800.0),
|
||||
])
|
||||
.unwrap();
|
||||
assert_eq!(equal_age.items[0].entity_id, 2);
|
||||
|
||||
let older_supported = run(vec![
|
||||
candidate(1, 0, 3.0, 0.0),
|
||||
candidate(2, 30 * 24, 40.0, 800.0),
|
||||
])
|
||||
.unwrap();
|
||||
assert_eq!(older_supported.items[0].entity_id, 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn components_explain_score_as_log_engagement_times_freshness() {
|
||||
let mut value = candidate(1, 48, 4.0, 100.0);
|
||||
value.signals.comments = 4.0;
|
||||
let result = run(vec![value]).unwrap();
|
||||
let item = &result.items[0];
|
||||
|
||||
assert!((item.components.engagement - 5.0).abs() < f64::EPSILON);
|
||||
assert!((item.components.response_rate - 7.0 / 120.0).abs() < f64::EPSILON);
|
||||
assert!((item.components.freshness - 0.5_f64.powf(0.25)).abs() < f64::EPSILON);
|
||||
assert_eq!(
|
||||
item.score,
|
||||
item.components.engagement.ln_1p() * item.components.freshness
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn output_is_a_bijection_of_the_input() {
|
||||
let candidates = vec![
|
||||
candidate(9, 100, 1.0, 20.0),
|
||||
candidate(2, 10, 50.0, 100.0),
|
||||
candidate(7, 0, 0.0, 1_000.0),
|
||||
];
|
||||
let expected = candidates
|
||||
.iter()
|
||||
.map(|candidate| candidate.entity_id)
|
||||
.collect::<BTreeSet<_>>();
|
||||
let result = run(candidates).unwrap();
|
||||
let actual = result
|
||||
.items
|
||||
.iter()
|
||||
.map(|item| item.entity_id)
|
||||
.collect::<BTreeSet<_>>();
|
||||
|
||||
assert_eq!(actual, expected);
|
||||
assert_eq!(result.items.len(), expected.len());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_and_maximum_candidate_sets_are_valid_but_over_limit_is_not() {
|
||||
assert!(run(Vec::new()).unwrap().items.is_empty());
|
||||
|
||||
let maximum = (0..MAX_EXACT_RANK_CANDIDATES as u64)
|
||||
.map(|entity_id| candidate(entity_id, 0, 0.0, 0.0))
|
||||
.collect();
|
||||
assert_eq!(run(maximum).unwrap().items.len(), MAX_EXACT_RANK_CANDIDATES);
|
||||
|
||||
let over_limit = (0..=MAX_EXACT_RANK_CANDIDATES as u64)
|
||||
.map(|entity_id| candidate(entity_id, 0, 0.0, 0.0))
|
||||
.collect();
|
||||
assert!(run(over_limit).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn duplicate_future_negative_and_non_finite_inputs_are_rejected() {
|
||||
assert!(run(vec![candidate(1, 0, 1.0, 1.0), candidate(1, 0, 2.0, 2.0)]).is_err());
|
||||
|
||||
let mut future = candidate(1, 0, 1.0, 1.0);
|
||||
future.created_at_nanos = AS_OF + 1;
|
||||
assert!(run(vec![future]).is_err());
|
||||
|
||||
for signals in [
|
||||
ExactRankSignals {
|
||||
hearts: -1.0,
|
||||
comments: 0.0,
|
||||
exposure: 0.0,
|
||||
},
|
||||
ExactRankSignals {
|
||||
hearts: 0.0,
|
||||
comments: f64::NAN,
|
||||
exposure: 0.0,
|
||||
},
|
||||
ExactRankSignals {
|
||||
hearts: 0.0,
|
||||
comments: 0.0,
|
||||
exposure: f64::INFINITY,
|
||||
},
|
||||
] {
|
||||
let mut invalid = candidate(1, 0, 0.0, 0.0);
|
||||
invalid.signals = signals;
|
||||
assert!(run(vec![invalid]).is_err());
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn exact_profile_name_and_version_are_pinned() {
|
||||
let mut wrong_name = request(Vec::new());
|
||||
wrong_name.profile = "other".into();
|
||||
assert!(execute(&wrong_name).is_err());
|
||||
|
||||
let mut wrong_version = request(Vec::new());
|
||||
wrong_version.profile_version = 2;
|
||||
assert!(execute(&wrong_version).is_err());
|
||||
|
||||
let response = execute(&request(Vec::new())).unwrap();
|
||||
assert_eq!(response.profile, EXACT_RANK_PROFILE);
|
||||
assert_eq!(response.profile_version, EXACT_RANK_PROFILE_VERSION);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn overflowing_computation_is_rejected() {
|
||||
let mut overflow = candidate(1, 0, f64::MAX, 0.0);
|
||||
overflow.signals.comments = f64::MAX;
|
||||
assert!(run(vec![overflow]).is_err());
|
||||
}
|
||||
@ -70,7 +70,7 @@ fn skeleton(name: &str) -> RankingProfile {
|
||||
|
||||
// ── Profile tuning constants ──────────────────────────────────────────────
|
||||
|
||||
/// Age-decay gravity for the hot sort (Reddit-style HN algorithm).
|
||||
/// Age-decay gravity for the view-count hot sort.
|
||||
/// Higher values decay older content faster.
|
||||
const HOT_GRAVITY: f64 = 1.8;
|
||||
|
||||
|
||||
@ -44,7 +44,7 @@ pub(super) const RELEVANCE_BASE_WEIGHT: f64 = 10.0;
|
||||
/// contribute on the same scale without overwhelming the base signal score.
|
||||
pub(super) const PREFERENCE_BOOST_WEIGHT: f64 = 0.3;
|
||||
|
||||
/// Hot: `log10(max(upvotes - downvotes, 1)) / (age_hours + 2)^gravity`
|
||||
/// Hot: `log10(max(views, 1)) / (age_hours + 2)^gravity`.
|
||||
pub(super) fn hot_score(views: f64, age_hours: f64, gravity: f64) -> f64 {
|
||||
views.max(1.0).log10() / (age_hours + 2.0).powf(gravity)
|
||||
}
|
||||
|
||||
181
tidal/tests/staged_signal_context.rs
Normal file
181
tidal/tests/staged_signal_context.rs
Normal file
@ -0,0 +1,181 @@
|
||||
//! The two-phase signal write must carry originating context.
|
||||
//!
|
||||
//! `signal_staged` applies `(signal, entity, weight)` and nothing else. The
|
||||
//! replicated cluster leader used it for every `POST /signals`, so a clustered
|
||||
//! deployment accepted each behavioural signal, answered `204`, and silently
|
||||
//! discarded `user_id`/`creator_id` — no hard negatives, no seen tracking, no
|
||||
//! interaction weight, no preference vector, and no wire evidence of the loss.
|
||||
//! `TidalDb::signal_with_context_staged` is the fix; these tests pin it.
|
||||
//!
|
||||
//! The shape is DIFFERENTIAL on purpose: each test drives the same signals
|
||||
//! through the synchronous `signal_with_context` and through the staged path,
|
||||
//! then asserts the two databases reach the same observable state. A staged
|
||||
//! path that drops context fails these on the assertions rather than needing a
|
||||
//! hand-written expected value, which is exactly the check the original bug got
|
||||
//! past.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use tidaldb::TidalDb;
|
||||
use tidaldb::schema::{DecaySpec, EntityKind, Schema, SchemaBuilder, Timestamp, Window};
|
||||
|
||||
const USER: u64 = 77;
|
||||
const CREATOR: u64 = 900;
|
||||
const ITEM: u64 = 4_242;
|
||||
|
||||
fn schema() -> Schema {
|
||||
let mut builder = SchemaBuilder::new();
|
||||
// `dislike` is a hard-negative signal; `like` is positive engagement. Both
|
||||
// branches of the context dispatch need coverage.
|
||||
for (name, half_life_days) in [("like", 14_u64), ("dislike", 1)] {
|
||||
let _ = builder
|
||||
.signal(
|
||||
name,
|
||||
EntityKind::Item,
|
||||
DecaySpec::Exponential {
|
||||
half_life: Duration::from_secs(half_life_days * 24 * 3600),
|
||||
},
|
||||
)
|
||||
.windows(&[Window::TwentyFourHours, Window::SevenDays, Window::AllTime])
|
||||
.velocity(true)
|
||||
.add();
|
||||
}
|
||||
builder.build().expect("schema must be valid")
|
||||
}
|
||||
|
||||
fn open_db() -> TidalDb {
|
||||
TidalDb::builder()
|
||||
.ephemeral()
|
||||
.with_schema(schema())
|
||||
.open()
|
||||
.expect("db open")
|
||||
}
|
||||
|
||||
/// `(is_negative, is_seen, interaction_score)` — the durable, user-scoped
|
||||
/// consequences of a context-carrying signal.
|
||||
fn observed(db: &TidalDb, now_ns: u64) -> (bool, bool, f64) {
|
||||
let item_slot = u32::try_from(ITEM).expect("test item id fits u32");
|
||||
(
|
||||
db.hard_negatives().is_negative(USER, item_slot),
|
||||
db.user_state().is_seen(USER, item_slot),
|
||||
db.interaction_ledger().score(USER, CREATOR, now_ns),
|
||||
)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn staged_context_matches_synchronous_for_hard_negative() {
|
||||
let ts = Timestamp::now();
|
||||
let sync_db = open_db();
|
||||
sync_db
|
||||
.signal_with_context("dislike", ITEM.into(), 1.0, ts, Some(USER), Some(CREATOR))
|
||||
.expect("synchronous context write");
|
||||
|
||||
let staged_db = open_db();
|
||||
staged_db
|
||||
.signal_with_context_staged("dislike", ITEM.into(), 1.0, ts, Some(USER), Some(CREATOR))
|
||||
.expect("stage")
|
||||
.wait(&staged_db)
|
||||
.expect("complete");
|
||||
|
||||
let now = ts.as_nanos();
|
||||
let sync = observed(&sync_db, now);
|
||||
let staged = observed(&staged_db, now);
|
||||
|
||||
// Positive control: the synchronous path really does record all three, so a
|
||||
// passing comparison cannot be two empty states agreeing with each other.
|
||||
assert!(sync.0, "synchronous dislike must record a hard negative");
|
||||
assert!(sync.1, "synchronous dislike must mark the item seen");
|
||||
assert!(
|
||||
sync.2 > 0.0,
|
||||
"synchronous dislike must record interaction weight"
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
staged.0, sync.0,
|
||||
"staged write dropped the hard negative — user_id did not reach the engine"
|
||||
);
|
||||
assert_eq!(
|
||||
staged.1, sync.1,
|
||||
"staged write dropped seen tracking — user_id did not reach the engine"
|
||||
);
|
||||
assert!(
|
||||
(staged.2 - sync.2).abs() < 1e-9,
|
||||
"staged write dropped interaction weight: staged={} sync={}",
|
||||
staged.2,
|
||||
sync.2
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn staged_context_matches_synchronous_for_positive_engagement() {
|
||||
let ts = Timestamp::now();
|
||||
let sync_db = open_db();
|
||||
sync_db
|
||||
.signal_with_context("like", ITEM.into(), 1.0, ts, Some(USER), Some(CREATOR))
|
||||
.expect("synchronous context write");
|
||||
|
||||
let staged_db = open_db();
|
||||
staged_db
|
||||
.signal_with_context_staged("like", ITEM.into(), 1.0, ts, Some(USER), Some(CREATOR))
|
||||
.expect("stage")
|
||||
.wait(&staged_db)
|
||||
.expect("complete");
|
||||
|
||||
let now = ts.as_nanos();
|
||||
let sync = observed(&sync_db, now);
|
||||
let staged = observed(&staged_db, now);
|
||||
|
||||
// A like is not a hard negative — pinned so a future dispatch change cannot
|
||||
// start hiding liked items without failing here.
|
||||
assert!(!sync.0, "a like must not record a hard negative");
|
||||
assert!(sync.1, "a like must mark the item seen");
|
||||
assert!(sync.2 > 0.0, "a like must record interaction weight");
|
||||
|
||||
assert_eq!(staged, sync, "staged like diverged from synchronous like");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn staged_write_without_context_still_records_the_base_signal() {
|
||||
let ts = Timestamp::now();
|
||||
let db = open_db();
|
||||
db.signal_with_context_staged("like", ITEM.into(), 1.0, ts, None, None)
|
||||
.expect("stage")
|
||||
.wait(&db)
|
||||
.expect("complete");
|
||||
|
||||
let item_slot = u32::try_from(ITEM).expect("test item id fits u32");
|
||||
// No context supplied ⇒ no user-scoped side effects, and in particular no
|
||||
// attribution to a user that was never named.
|
||||
assert!(!db.user_state().is_seen(USER, item_slot));
|
||||
assert!(!db.hard_negatives().is_negative(USER, item_slot));
|
||||
assert!(
|
||||
db.read_decay_score(ITEM.into(), "like", 0)
|
||||
.expect("read score")
|
||||
.is_some_and(|score| score > 0.0),
|
||||
"the base signal must still be recorded without context"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn staged_context_rejects_an_item_id_past_the_u32_universe() {
|
||||
let db = open_db();
|
||||
// The u32 item-slot guard must fire at STAGING, before anything is written —
|
||||
// a durable Tag::HardNeg row keyed on a truncated id is a permanent
|
||||
// cross-item collision that survives restart.
|
||||
let over_range = u64::from(u32::MAX) + 1;
|
||||
let err = db
|
||||
.signal_with_context_staged(
|
||||
"dislike",
|
||||
over_range.into(),
|
||||
1.0,
|
||||
Timestamp::now(),
|
||||
Some(USER),
|
||||
None,
|
||||
)
|
||||
.expect_err("an over-range item id with a user context must be rejected");
|
||||
let message = err.to_string();
|
||||
assert!(
|
||||
message.contains("u32 item-universe limit"),
|
||||
"unexpected error: {message}"
|
||||
);
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user