Reconciles two independently-developed lines from base 006d3d0:
ours — M9/M10 community layers, retroactive purge + re-materialization,
signal revocation, agent capability boundaries, P1 feedback loop,
reason labels, instrumented metrics
theirs — M11/M12 cluster mode (tidal-net gRPC transport, tidal-server
cluster/scatter-gather, tidal-stress), multi-vector preference,
ANN candidate-gen, warm-tier day buckets, keyed signal snapshots
Notable semantic resolutions:
* storage::keys::Tag — both sides allocated 0x0E..0x11 for different
records. Kept theirs' 0x0E..0x1A (shipped on-disk format) and renumbered
ours to 0x1B..0x1E (CommunityMembership/Revocation/PurgeManifest/
CommunityLeave); Tag::ALL grown to 30 so the contiguity drift guard holds.
* ranking executor — took theirs' rewrite (SignalReadPlan pre-pass, keyed
SignalKey snapshots, Result-returning reads, finalize()) and re-applied
ours' M10 read suppression at the chokepoints it introduced:
single_signal_score, score_hot/trending/controversial,
CreatorEngagementRate, and the Stage-4 boost loop.
* signals::warm — theirs' day-bucket/read-time-rotation rewrite, with ours'
subtract_bucket and Clone extended to the new day tier; ours' test split
kept (warm/tests.rs, warm/proptests.rs) carrying theirs' updated bodies.
* db::signals — kept ours' contribution-logging try_cohort_attribution in
signal_dispatch.rs and theirs' event-time try_update_preference_vector;
dropped the superseded duplicates.
* db::mod / from_parts — theirs' constructors, with ours' purge/
re-materialization/revocation/community/skip-counter fields and restart
rebuilds; from_parts kept in its own file per the 600-line guideline.
* schema::validation::builders — ours' module split with theirs' expanded
tests; policy validation runs both sides' checks (read-signal lists +
profile overrides, then the zero-duration limit guard).
* feedback Unhide no longer writes a -1.0 "hide" signal: theirs' engine
rejects negative weights (spec §8). Reverses index state only, matching
every other undo action.
* SessionState::new is now the single construction path (gains
overrides_rejected/default_profile); AuditEntry gains kind on the
deserialize path, inferred from the accepted flag as before.
* Removed tidal/src/replication/tcp_transport.rs and its test: never
declared in replication/mod.rs on either branch, so it had never
compiled and nothing referenced it. Superseded by tidal-net's
GrpcTransport.
Verified: cargo clippy -p tidaldb (lib) clean; --all-targets compiles for
tidaldb/tidal-net/tidal-server/tidal-stress; 2094/2094 lib tests and the
integration suite pass except m8p3_reconcile_production's two CRDT-count
assertions, which fail identically on MERGE_HEAD (pre-existing).
tidalctl cannot build locally: its aws-sdk deps need rustc 1.91.1, local
toolchain is 1.91.0.
427 lines
12 KiB
Rust
427 lines
12 KiB
Rust
//! Integration tests for M7P4 Operational Visibility — session metrics + cross-session aggregation.
|
|
//!
|
|
//! - Task 04: Session + cohort + degradation metrics
|
|
//! - Task 09: Cross-session aggregation
|
|
#![allow(clippy::too_many_lines, clippy::unwrap_used)]
|
|
|
|
use std::{collections::HashMap, time::Duration};
|
|
|
|
use tidaldb::{
|
|
TidalDb,
|
|
schema::{DecaySpec, EntityId, EntityKind, Timestamp, Window},
|
|
};
|
|
|
|
// ── Prometheus output helpers ────────────────────────────────────────────────
|
|
|
|
/// Extract the numeric value for a metric from Prometheus text output.
|
|
///
|
|
/// Looks for a line that starts with `name` (not a comment), then parses
|
|
/// the trailing float. Returns `None` if the metric is absent.
|
|
#[cfg(feature = "metrics")]
|
|
fn prometheus_value(prom: &str, name: &str) -> Option<f64> {
|
|
prom.lines()
|
|
.filter(|l| !l.starts_with('#'))
|
|
.find(|l| l.starts_with(name) && l[name.len()..].starts_with(' '))
|
|
.and_then(|l| l.split_whitespace().last())
|
|
.and_then(|v| v.parse().ok())
|
|
}
|
|
|
|
// ── Shared test schema ───────────────────────────────────────────────────────
|
|
|
|
fn build_test_schema() -> tidaldb::schema::Schema {
|
|
use tidaldb::{AgentPolicy, schema::SchemaBuilder};
|
|
|
|
let mut builder = SchemaBuilder::new();
|
|
let _ = builder
|
|
.signal(
|
|
"view",
|
|
EntityKind::Item,
|
|
DecaySpec::Exponential {
|
|
half_life: Duration::from_secs(7 * 24 * 3600),
|
|
},
|
|
)
|
|
.windows(&[Window::OneHour])
|
|
.add();
|
|
let _ = builder
|
|
.signal(
|
|
"like",
|
|
EntityKind::Item,
|
|
DecaySpec::Exponential {
|
|
half_life: Duration::from_secs(30 * 24 * 3600),
|
|
},
|
|
)
|
|
.windows(&[Window::OneHour])
|
|
.add();
|
|
builder.session_policy(
|
|
"default",
|
|
AgentPolicy {
|
|
allowed_signals: vec!["view".to_string(), "like".to_string()],
|
|
denied_signals: vec![],
|
|
max_session_duration: Duration::from_secs(3600),
|
|
max_signals_per_session: 1000,
|
|
..AgentPolicy::default()
|
|
},
|
|
);
|
|
builder.build().unwrap()
|
|
}
|
|
|
|
// ── Task 04: Session + Cohort + Degradation Metrics ─────────────────────────
|
|
|
|
#[cfg(feature = "metrics")]
|
|
#[test]
|
|
fn session_start_increments_active_sessions() {
|
|
let schema = build_test_schema();
|
|
let db = TidalDb::builder()
|
|
.ephemeral()
|
|
.with_schema(schema)
|
|
.open()
|
|
.unwrap();
|
|
|
|
let before = prometheus_value(&db.metrics().render_prometheus(), "tidaldb_active_sessions")
|
|
.unwrap_or(0.0);
|
|
let meta = HashMap::new();
|
|
let _handle = db.start_session(1u64, "agent1", "default", meta).unwrap();
|
|
let after =
|
|
prometheus_value(&db.metrics().render_prometheus(), "tidaldb_active_sessions").unwrap();
|
|
|
|
assert!(
|
|
((after - before) - 1.0).abs() < 0.5,
|
|
"active_sessions must increment when a session starts"
|
|
);
|
|
}
|
|
|
|
#[cfg(feature = "metrics")]
|
|
#[test]
|
|
fn session_close_updates_active_and_closed_counters() {
|
|
let schema = build_test_schema();
|
|
let db = TidalDb::builder()
|
|
.ephemeral()
|
|
.with_schema(schema)
|
|
.open()
|
|
.unwrap();
|
|
|
|
let before_active =
|
|
prometheus_value(&db.metrics().render_prometheus(), "tidaldb_active_sessions")
|
|
.unwrap_or(0.0);
|
|
let before_closed = prometheus_value(
|
|
&db.metrics().render_prometheus(),
|
|
"tidaldb_closed_sessions_total",
|
|
)
|
|
.unwrap_or(0.0);
|
|
|
|
let meta = HashMap::new();
|
|
let handle = db
|
|
.start_session(42u64, "agent-close-test", "default", meta)
|
|
.unwrap();
|
|
db.close_session(handle).unwrap();
|
|
|
|
let after_active =
|
|
prometheus_value(&db.metrics().render_prometheus(), "tidaldb_active_sessions").unwrap();
|
|
let after_closed = prometheus_value(
|
|
&db.metrics().render_prometheus(),
|
|
"tidaldb_closed_sessions_total",
|
|
)
|
|
.unwrap();
|
|
|
|
assert!(
|
|
(after_active - before_active).abs() < 0.5,
|
|
"active_sessions must return to its previous value after close_session"
|
|
);
|
|
assert!(
|
|
((after_closed - before_closed) - 1.0).abs() < 0.5,
|
|
"closed_sessions_total must increment on close_session"
|
|
);
|
|
}
|
|
|
|
#[cfg(feature = "metrics")]
|
|
#[test]
|
|
fn degradation_level_appears_in_prometheus_output() {
|
|
let schema = build_test_schema();
|
|
let db = TidalDb::builder()
|
|
.ephemeral()
|
|
.with_schema(schema)
|
|
.open()
|
|
.unwrap();
|
|
|
|
let prom = db.metrics().render_prometheus();
|
|
|
|
// Default: Full fidelity (level 0).
|
|
let level = prometheus_value(&prom, "tidaldb_degradation_level").unwrap();
|
|
assert!(
|
|
level.abs() < 0.5,
|
|
"initial degradation level must be 0 (Full fidelity)"
|
|
);
|
|
|
|
assert!(
|
|
prom.contains("tidaldb_degradation_level"),
|
|
"Prometheus output must include degradation_level gauge"
|
|
);
|
|
assert!(
|
|
prom.contains("tidaldb_rate_limited_total"),
|
|
"Prometheus output must include rate_limited_total counter"
|
|
);
|
|
}
|
|
|
|
// ── Task 09: Cross-Session Aggregation ──────────────────────────────────────
|
|
|
|
#[test]
|
|
fn user_session_summary_aggregates_correctly() {
|
|
let schema = build_test_schema();
|
|
let db = TidalDb::builder()
|
|
.ephemeral()
|
|
.with_schema(schema)
|
|
.open()
|
|
.unwrap();
|
|
let user_id = 42u64;
|
|
let meta = HashMap::new();
|
|
|
|
// Run 3 sessions, each with 2 views + 1 like = 3 signals per session.
|
|
for i in 0u64..3 {
|
|
let handle = db
|
|
.start_session(user_id, "test", "default", meta.clone())
|
|
.unwrap();
|
|
db.session_signal(
|
|
&handle,
|
|
"view",
|
|
EntityId::new(i * 10 + 1),
|
|
1.0,
|
|
Timestamp::now(),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
db.session_signal(
|
|
&handle,
|
|
"view",
|
|
EntityId::new(i * 10 + 2),
|
|
1.0,
|
|
Timestamp::now(),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
db.session_signal(
|
|
&handle,
|
|
"like",
|
|
EntityId::new(i * 10 + 3),
|
|
1.0,
|
|
Timestamp::now(),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
db.close_session(handle).unwrap();
|
|
}
|
|
|
|
let summary = db.user_session_summary(user_id, 0).unwrap();
|
|
assert_eq!(summary.sessions_count, 3);
|
|
assert_eq!(summary.total_signals, 9);
|
|
assert_eq!(summary.total_rejections, 0);
|
|
assert_eq!(summary.user_id, user_id);
|
|
assert_eq!(summary.since_ns, 0);
|
|
assert!(summary.earliest_session_ns.is_some());
|
|
assert!(summary.latest_session_ns.is_some());
|
|
|
|
// Top signal types: "view" first (6 total), "like" second (3 total).
|
|
assert_eq!(summary.top_signal_types.len(), 2);
|
|
assert_eq!(summary.top_signal_types[0].0, "view");
|
|
assert_eq!(summary.top_signal_types[0].1, 6);
|
|
assert_eq!(summary.top_signal_types[1].0, "like");
|
|
assert_eq!(summary.top_signal_types[1].1, 3);
|
|
}
|
|
|
|
#[test]
|
|
fn user_session_summary_no_sessions_returns_not_found() {
|
|
let schema = build_test_schema();
|
|
let db = TidalDb::builder()
|
|
.ephemeral()
|
|
.with_schema(schema)
|
|
.open()
|
|
.unwrap();
|
|
|
|
let result = db.user_session_summary(999, 0);
|
|
assert!(result.is_err());
|
|
assert!(matches!(result, Err(tidaldb::TidalError::NotFound { .. })));
|
|
}
|
|
|
|
#[test]
|
|
fn user_session_summary_different_user_excluded() {
|
|
let schema = build_test_schema();
|
|
let db = TidalDb::builder()
|
|
.ephemeral()
|
|
.with_schema(schema)
|
|
.open()
|
|
.unwrap();
|
|
let meta = HashMap::new();
|
|
|
|
// User 1: one view signal.
|
|
let handle = db
|
|
.start_session(1u64, "test", "default", meta.clone())
|
|
.unwrap();
|
|
db.session_signal(
|
|
&handle,
|
|
"view",
|
|
EntityId::new(1),
|
|
1.0,
|
|
Timestamp::now(),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
db.close_session(handle).unwrap();
|
|
|
|
// User 2: one like signal.
|
|
let handle = db.start_session(2u64, "test", "default", meta).unwrap();
|
|
db.session_signal(
|
|
&handle,
|
|
"like",
|
|
EntityId::new(2),
|
|
1.0,
|
|
Timestamp::now(),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
db.close_session(handle).unwrap();
|
|
|
|
// Query user 1 only — user 2's session must be excluded.
|
|
let summary = db.user_session_summary(1, 0).unwrap();
|
|
assert_eq!(summary.sessions_count, 1);
|
|
assert_eq!(summary.total_signals, 1);
|
|
assert_eq!(summary.top_signal_types.len(), 1);
|
|
assert_eq!(summary.top_signal_types[0].0, "view");
|
|
assert_eq!(summary.top_signal_types[0].1, 1);
|
|
}
|
|
|
|
#[test]
|
|
fn user_session_summary_since_ns_filter() {
|
|
let schema = build_test_schema();
|
|
let db = TidalDb::builder()
|
|
.ephemeral()
|
|
.with_schema(schema)
|
|
.open()
|
|
.unwrap();
|
|
let user_id = 100u64;
|
|
let meta = HashMap::new();
|
|
|
|
// Session 1.
|
|
let handle = db
|
|
.start_session(user_id, "test", "default", meta.clone())
|
|
.unwrap();
|
|
db.session_signal(
|
|
&handle,
|
|
"view",
|
|
EntityId::new(1),
|
|
1.0,
|
|
Timestamp::now(),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
db.close_session(handle).unwrap();
|
|
|
|
// Record time between sessions.
|
|
let midpoint = Timestamp::now().as_nanos();
|
|
|
|
// Session 2 (after midpoint).
|
|
let handle = db.start_session(user_id, "test", "default", meta).unwrap();
|
|
db.session_signal(
|
|
&handle,
|
|
"like",
|
|
EntityId::new(2),
|
|
1.0,
|
|
Timestamp::now(),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
db.close_session(handle).unwrap();
|
|
|
|
// since_ns = midpoint must exclude session 1 and include only session 2.
|
|
let summary = db.user_session_summary(user_id, midpoint).unwrap();
|
|
assert_eq!(summary.sessions_count, 1);
|
|
assert_eq!(summary.total_signals, 1);
|
|
assert_eq!(summary.top_signal_types[0].0, "like");
|
|
}
|
|
|
|
#[test]
|
|
fn session_snapshot_has_timing_fields() {
|
|
let schema = build_test_schema();
|
|
let db = TidalDb::builder()
|
|
.ephemeral()
|
|
.with_schema(schema)
|
|
.open()
|
|
.unwrap();
|
|
let user_id = 50u64;
|
|
let meta = HashMap::new();
|
|
|
|
let before = Timestamp::now().as_nanos();
|
|
|
|
let handle = db.start_session(user_id, "test", "default", meta).unwrap();
|
|
db.session_signal(
|
|
&handle,
|
|
"view",
|
|
EntityId::new(1),
|
|
1.0,
|
|
Timestamp::now(),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
|
|
// Active snapshot: started_at_ns populated, closed_at_ns = 0.
|
|
let active_snap = db.session_snapshot(handle.id).unwrap();
|
|
assert!(
|
|
active_snap.started_at_ns >= before,
|
|
"started_at_ns should be after test start"
|
|
);
|
|
assert_eq!(
|
|
active_snap.closed_at_ns, 0,
|
|
"active session should have closed_at_ns = 0"
|
|
);
|
|
|
|
db.close_session(handle).unwrap();
|
|
|
|
let after = Timestamp::now().as_nanos();
|
|
|
|
// Summary timing should bracket the session.
|
|
let summary = db.user_session_summary(user_id, 0).unwrap();
|
|
assert!(summary.earliest_session_ns.unwrap() >= before);
|
|
assert!(summary.latest_session_ns.unwrap() <= after);
|
|
}
|
|
|
|
#[test]
|
|
fn signal_snap_entry_has_count() {
|
|
let schema = build_test_schema();
|
|
let db = TidalDb::builder()
|
|
.ephemeral()
|
|
.with_schema(schema)
|
|
.open()
|
|
.unwrap();
|
|
let user_id = 60u64;
|
|
let meta = HashMap::new();
|
|
|
|
let handle = db.start_session(user_id, "test", "default", meta).unwrap();
|
|
// Write 3 view signals and 1 like signal.
|
|
for entity in [1u64, 2, 3] {
|
|
db.session_signal(
|
|
&handle,
|
|
"view",
|
|
EntityId::new(entity),
|
|
1.0,
|
|
Timestamp::now(),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
}
|
|
db.session_signal(
|
|
&handle,
|
|
"like",
|
|
EntityId::new(4),
|
|
1.0,
|
|
Timestamp::now(),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
|
|
let snap = db.session_snapshot(handle.id).unwrap();
|
|
let view_entry = snap.signals.get("view").unwrap();
|
|
assert_eq!(view_entry.count, 3);
|
|
let like_entry = snap.signals.get("like").unwrap();
|
|
assert_eq!(like_entry.count, 1);
|
|
|
|
db.close_session(handle).unwrap();
|
|
}
|