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.
395 lines
12 KiB
Rust
395 lines
12 KiB
Rust
//! M9 Integration Tests: Leave & Stop-Forward (`m9-leave-revocation`).
|
|
//!
|
|
//! Validates that:
|
|
//! - `leave_community_layer` immediately stops signal fan-out to cohort aggregates.
|
|
//! - `rejoin_community_layer` restores fan-out.
|
|
//! - Both calls are idempotent.
|
|
//! - The leave timestamp is durable across restarts.
|
|
//! - Unknown users default to `Active`.
|
|
//!
|
|
//! Run with:
|
|
//! ```bash
|
|
//! cargo test --manifest-path tidal/Cargo.toml --test m9_leave_revocation
|
|
//! ```
|
|
|
|
#![allow(clippy::unwrap_used, clippy::cast_precision_loss)]
|
|
|
|
use std::collections::HashMap;
|
|
use std::time::Duration;
|
|
|
|
use tidaldb::TidalDb;
|
|
use tidaldb::cohort::{CohortDef, Predicate};
|
|
use tidaldb::entities::MembershipStatus;
|
|
use tidaldb::schema::{DecaySpec, EntityId, EntityKind, SchemaBuilder, Timestamp, Window};
|
|
|
|
// ── Schema ───────────────────────────────────────────────────────────────────
|
|
|
|
fn leave_schema() -> tidaldb::schema::Schema {
|
|
let mut builder = SchemaBuilder::new();
|
|
let _ = builder
|
|
.signal(
|
|
"view",
|
|
EntityKind::Item,
|
|
DecaySpec::Exponential {
|
|
half_life: Duration::from_secs(7 * 24 * 3600),
|
|
},
|
|
)
|
|
.windows(&[Window::AllTime])
|
|
.velocity(false)
|
|
.add();
|
|
builder.build().expect("leave schema must be valid")
|
|
}
|
|
|
|
// ── Helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
/// Open an ephemeral db with a single always-true cohort "test-cohort".
|
|
fn open_db_with_cohort() -> TidalDb {
|
|
let db = TidalDb::builder()
|
|
.ephemeral()
|
|
.with_schema(leave_schema())
|
|
.open()
|
|
.expect("db open");
|
|
|
|
// Always-true cohort: matches any user that has "member" = "yes".
|
|
db.define_cohort(CohortDef {
|
|
name: "test-cohort".to_string(),
|
|
predicate: Predicate::Eq {
|
|
field: "member".into(),
|
|
value: "yes".into(),
|
|
},
|
|
})
|
|
.expect("define cohort");
|
|
db
|
|
}
|
|
|
|
const USER_ID: u64 = 1001;
|
|
const ITEM_ID: u64 = 42;
|
|
const CREATOR_ID: u64 = 99;
|
|
|
|
/// Write user metadata so they match the test cohort.
|
|
fn register_user(db: &TidalDb) {
|
|
let mut meta = HashMap::new();
|
|
meta.insert("member".to_string(), "yes".to_string());
|
|
db.write_user(EntityId::new(USER_ID), &meta).unwrap();
|
|
}
|
|
|
|
/// Write one view signal with context for USER_ID and return the cohort ledger AllTime count.
|
|
fn signal_and_count(db: &TidalDb) -> u64 {
|
|
let ts = Timestamp::now();
|
|
db.signal_with_context(
|
|
"view",
|
|
EntityId::new(ITEM_ID),
|
|
1.0,
|
|
ts,
|
|
Some(USER_ID),
|
|
Some(CREATOR_ID),
|
|
)
|
|
.unwrap();
|
|
db.cohort_ledger()
|
|
.read_windowed_count(
|
|
"test-cohort",
|
|
EntityId::new(ITEM_ID),
|
|
"view",
|
|
Window::AllTime,
|
|
)
|
|
.unwrap()
|
|
}
|
|
|
|
// ── TC-01: leave stops cohort fan-out ────────────────────────────────────────
|
|
|
|
#[test]
|
|
fn leave_stops_cohort_fanout() {
|
|
let db = open_db_with_cohort();
|
|
register_user(&db);
|
|
|
|
// Baseline: signal before leave — cohort count increases.
|
|
let count_before = signal_and_count(&db);
|
|
assert!(
|
|
count_before > 0,
|
|
"cohort should receive signals before leave"
|
|
);
|
|
|
|
// Leave.
|
|
db.leave_community_layer(USER_ID).unwrap();
|
|
|
|
// Post-leave signal: cohort count must NOT increase.
|
|
let count_after = signal_and_count(&db);
|
|
assert_eq!(
|
|
count_after, count_before,
|
|
"cohort count must not increase after leave"
|
|
);
|
|
|
|
// Base signal ledger IS still updated (entity decay score).
|
|
let decay = db
|
|
.read_decay_score(EntityId::new(ITEM_ID), "view", 0)
|
|
.unwrap();
|
|
assert!(
|
|
decay.is_some(),
|
|
"base entity decay score must still be recorded after leave"
|
|
);
|
|
}
|
|
|
|
// ── TC-02: rejoin resumes cohort fan-out ─────────────────────────────────────
|
|
|
|
#[test]
|
|
fn join_resumes_cohort_fanout() {
|
|
let db = open_db_with_cohort();
|
|
register_user(&db);
|
|
|
|
// Leave.
|
|
db.leave_community_layer(USER_ID).unwrap();
|
|
|
|
// Confirm fan-out is stopped.
|
|
let count_left = signal_and_count(&db);
|
|
|
|
// Rejoin.
|
|
db.rejoin_community_layer(USER_ID).unwrap();
|
|
|
|
// Fan-out should resume.
|
|
let count_rejoined = signal_and_count(&db);
|
|
assert!(
|
|
count_rejoined > count_left,
|
|
"cohort count must increase after rejoin: count_left={count_left}, count_rejoined={count_rejoined}"
|
|
);
|
|
}
|
|
|
|
// ── TC-03: leave is idempotent ────────────────────────────────────────────────
|
|
|
|
#[test]
|
|
fn leave_is_idempotent() {
|
|
let db = open_db_with_cohort();
|
|
|
|
// First leave.
|
|
db.leave_community_layer(USER_ID).unwrap();
|
|
let status_1 = db.community_layer_status(USER_ID).unwrap();
|
|
assert_eq!(
|
|
status_1.status,
|
|
MembershipStatus::Left,
|
|
"status must be Left after first leave"
|
|
);
|
|
let left_at_1 = status_1.left_at_ns.expect("left_at_ns must be set");
|
|
|
|
// Second leave: must succeed and update timestamp.
|
|
db.leave_community_layer(USER_ID).unwrap();
|
|
let status_2 = db.community_layer_status(USER_ID).unwrap();
|
|
assert_eq!(
|
|
status_2.status,
|
|
MembershipStatus::Left,
|
|
"status must remain Left after second leave"
|
|
);
|
|
let left_at_2 = status_2
|
|
.left_at_ns
|
|
.expect("left_at_ns must be set after second leave");
|
|
|
|
// Timestamp is updated (not corrupted) — second call is >= first.
|
|
assert!(
|
|
left_at_2 >= left_at_1,
|
|
"second left_at_ns={left_at_2} must be >= first left_at_ns={left_at_1}"
|
|
);
|
|
}
|
|
|
|
// ── TC-04: status query lifecycle ────────────────────────────────────────────
|
|
|
|
#[test]
|
|
fn status_query_lifecycle() {
|
|
let db = open_db_with_cohort();
|
|
|
|
// 1. Fresh user — defaults to Active with no left_at_ns.
|
|
let initial = db.community_layer_status(USER_ID).unwrap();
|
|
assert_eq!(
|
|
initial.status,
|
|
MembershipStatus::Active,
|
|
"fresh user must default to Active"
|
|
);
|
|
assert!(
|
|
initial.left_at_ns.is_none(),
|
|
"fresh user must have no left_at_ns"
|
|
);
|
|
|
|
// 2. Leave — status becomes Left with a timestamp.
|
|
db.leave_community_layer(USER_ID).unwrap();
|
|
let after_leave = db.community_layer_status(USER_ID).unwrap();
|
|
assert_eq!(after_leave.status, MembershipStatus::Left);
|
|
assert!(
|
|
after_leave.left_at_ns.is_some(),
|
|
"left_at_ns must be set after leave"
|
|
);
|
|
|
|
// 3. Rejoin — status becomes Active; left_at_ns history is preserved.
|
|
db.rejoin_community_layer(USER_ID).unwrap();
|
|
let after_rejoin = db.community_layer_status(USER_ID).unwrap();
|
|
assert_eq!(after_rejoin.status, MembershipStatus::Active);
|
|
assert!(
|
|
after_rejoin.left_at_ns.is_some(),
|
|
"left_at_ns must be preserved after rejoin"
|
|
);
|
|
}
|
|
|
|
// ── TC-05: left_at_ns preserved on rejoin ────────────────────────────────────
|
|
|
|
#[test]
|
|
fn left_at_ns_preserved_on_rejoin() {
|
|
let db = open_db_with_cohort();
|
|
|
|
db.leave_community_layer(USER_ID).unwrap();
|
|
let after_leave = db.community_layer_status(USER_ID).unwrap();
|
|
let original_ts = after_leave.left_at_ns.expect("must have left_at_ns");
|
|
|
|
db.rejoin_community_layer(USER_ID).unwrap();
|
|
let after_rejoin = db.community_layer_status(USER_ID).unwrap();
|
|
|
|
assert_eq!(
|
|
after_rejoin.status,
|
|
MembershipStatus::Active,
|
|
"status must be Active after rejoin"
|
|
);
|
|
let preserved_ts = after_rejoin
|
|
.left_at_ns
|
|
.expect("left_at_ns must be preserved after rejoin");
|
|
assert_eq!(
|
|
preserved_ts, original_ts,
|
|
"left_at_ns must not change on rejoin"
|
|
);
|
|
}
|
|
|
|
// ── TC-06: durability across reopen ──────────────────────────────────────────
|
|
|
|
#[test]
|
|
#[cfg(feature = "test-utils")]
|
|
fn durability_across_reopen() {
|
|
use tidaldb::TempTidalHome;
|
|
|
|
let home = TempTidalHome::new().unwrap();
|
|
|
|
let original_left_at_ns: u64;
|
|
|
|
// Open, leave, close.
|
|
{
|
|
let db = TidalDb::builder()
|
|
.with_data_dir(home.path())
|
|
.with_schema(leave_schema())
|
|
.open()
|
|
.unwrap();
|
|
|
|
db.leave_community_layer(USER_ID).unwrap();
|
|
let status = db.community_layer_status(USER_ID).unwrap();
|
|
assert_eq!(status.status, MembershipStatus::Left);
|
|
original_left_at_ns = status.left_at_ns.expect("must have left_at_ns");
|
|
|
|
db.close().unwrap();
|
|
}
|
|
|
|
// Reopen: leave status must be restored.
|
|
{
|
|
let db = TidalDb::builder()
|
|
.with_data_dir(home.path())
|
|
.with_schema(leave_schema())
|
|
.open()
|
|
.unwrap();
|
|
|
|
let status = db.community_layer_status(USER_ID).unwrap();
|
|
assert_eq!(
|
|
status.status,
|
|
MembershipStatus::Left,
|
|
"leave status must survive restart"
|
|
);
|
|
let restored_ts = status.left_at_ns.expect("left_at_ns must survive restart");
|
|
assert_eq!(
|
|
restored_ts, original_left_at_ns,
|
|
"left_at_ns must be identical after restart"
|
|
);
|
|
|
|
db.close().unwrap();
|
|
}
|
|
}
|
|
|
|
// ── TC-07: unknown user defaults to active ───────────────────────────────────
|
|
|
|
#[test]
|
|
fn unknown_user_defaults_to_active() {
|
|
let db = open_db_with_cohort();
|
|
|
|
// Query a user that has never had any interaction.
|
|
let never_seen_user: u64 = 99_999_999;
|
|
let status = db.community_layer_status(never_seen_user).unwrap();
|
|
|
|
assert_eq!(
|
|
status.status,
|
|
MembershipStatus::Active,
|
|
"unknown user must default to Active"
|
|
);
|
|
assert!(
|
|
status.left_at_ns.is_none(),
|
|
"unknown user must have no left_at_ns"
|
|
);
|
|
}
|
|
|
|
// ── Regression: community forwarding also gated ───────────────────────────────
|
|
|
|
/// Verify that `try_community_forwarding` is also gated on leave status.
|
|
///
|
|
/// The `community_membership` opt-in (separate from the leave gate) is used for
|
|
/// community forwarding. This test confirms that leaving the community layer
|
|
/// stops forwarding even when the user is opted into a community membership.
|
|
#[test]
|
|
fn leave_stops_community_forwarding() {
|
|
let db = open_db_with_cohort();
|
|
register_user(&db);
|
|
|
|
// Opt user into community membership (separate from cohort registry).
|
|
db.join_community_layer(USER_ID, "test-cohort").unwrap();
|
|
|
|
// Write one signal before leave to establish baseline.
|
|
let ts = Timestamp::now();
|
|
db.signal_with_context(
|
|
"view",
|
|
EntityId::new(ITEM_ID),
|
|
1.0,
|
|
ts,
|
|
Some(USER_ID),
|
|
Some(CREATOR_ID),
|
|
)
|
|
.unwrap();
|
|
|
|
let community_key = "community::test-cohort".to_string();
|
|
let count_before = db
|
|
.cohort_ledger()
|
|
.read_windowed_count(
|
|
&community_key,
|
|
EntityId::new(ITEM_ID),
|
|
"view",
|
|
Window::AllTime,
|
|
)
|
|
.unwrap_or(0);
|
|
|
|
// Leave the community layer.
|
|
db.leave_community_layer(USER_ID).unwrap();
|
|
|
|
// Write another signal — should not forward to community aggregate.
|
|
let ts2 = Timestamp::now();
|
|
db.signal_with_context(
|
|
"view",
|
|
EntityId::new(ITEM_ID),
|
|
1.0,
|
|
ts2,
|
|
Some(USER_ID),
|
|
Some(CREATOR_ID),
|
|
)
|
|
.unwrap();
|
|
|
|
let count_after = db
|
|
.cohort_ledger()
|
|
.read_windowed_count(
|
|
&community_key,
|
|
EntityId::new(ITEM_ID),
|
|
"view",
|
|
Window::AllTime,
|
|
)
|
|
.unwrap_or(0);
|
|
|
|
assert_eq!(
|
|
count_after, count_before,
|
|
"community forwarding must be stopped after leave"
|
|
);
|
|
}
|