tidaldb/tidal/tests/review_pass2_d_replication.rs

//! Review pass-2, Zone D (Replication + tidal-net transport) regression tests.
//!
//! W8: `signal_for_tenant` must fail CLOSED *before any write* when a dual-write
//! resolves a non-local shard, so a caller that retries the returned error
//! cannot double-count the local half (signal accumulation is not idempotent).
//! The companion case proves a fully-local resolution still writes.

#![allow(clippy::unwrap_used)]

use std::time::Duration;

use tidaldb::{
    TidalDb, TidalError,
    replication::{RegionId, ShardAssignment, ShardId, TenantId},
    schema::{DecaySpec, EntityId, EntityKind, SchemaBuilder, Timestamp, Window},
};

fn open_db_with_view_signal() -> TidalDb {
    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();
    let schema = builder.build().unwrap();

    TidalDb::builder()
        .ephemeral()
        .with_schema(schema)
        .open()
        .expect("ephemeral db opens")
}

/// W8: a dual-write that resolves a non-local shard must fail closed with ZERO
/// local writes — the previous behavior wrote the local half and then errored,
/// so a retry double-counted the signal. After the error, the entity must have
/// NO decay score, proving the local half was never applied.
#[test]
fn signal_for_tenant_remote_shard_fails_closed_with_zero_writes() {
    let db = open_db_with_view_signal();

    // This node is the default local shard (ShardId::SINGLE = 0). Add a second
    // shard and put the default tenant into dual-write so a write resolves BOTH
    // the local shard 0 and the non-local shard 1.
    db.control_plane().update_topology(ShardAssignment {
        shard_id: ShardId(1),
        region_id: RegionId(1),
    });
    db.tenant_router()
        .set_dual_write(TenantId::DEFAULT, ShardId(0), ShardId(1));

    let item = EntityId::new(42);
    let err = db
        .signal_for_tenant(TenantId::DEFAULT, "view", item, 1.0, Timestamp::now())
        .expect_err("remote-shard dual-write must fail closed, not silently drop");
    assert!(
        matches!(err, TidalError::Internal(_)),
        "expected Internal (fail-closed) error, got: {err}"
    );

    // The local half must NOT have been applied: a retry of the Err cannot
    // double-count because nothing was written. The entity has no decay score.
    let score = db
        .read_decay_score(item, "view", 0)
        .expect("read local decay score");
    assert!(
        score.is_none() || score == Some(0.0),
        "fail-closed must write nothing locally; got score {score:?}"
    );

    // Retrying the failed op repeatedly must still leave nothing written — the
    // exact retry-amplification the W8 fix removes.
    for _ in 0..5 {
        let _ = db.signal_for_tenant(TenantId::DEFAULT, "view", item, 1.0, Timestamp::now());
    }
    let score_after_retries = db
        .read_decay_score(item, "view", 0)
        .expect("read local decay score after retries");
    assert!(
        score_after_retries.is_none() || score_after_retries == Some(0.0),
        "retries of a fail-closed op must not accumulate any local weight; got {score_after_retries:?}"
    );
}

/// Companion: when every resolved assignment is local (normal, non-migrating
/// tenant) the local write still lands — fail-closed only triggers on a
/// non-local assignment.
#[test]
fn signal_for_tenant_all_local_writes_succeed() {
    let db = open_db_with_view_signal();

    let item = EntityId::new(7);
    db.signal_for_tenant(TenantId::DEFAULT, "view", item, 1.0, Timestamp::now())
        .expect("a fully-local tenant write must succeed");

    let score = db
        .read_decay_score(item, "view", 0)
        .expect("read local decay score")
        .expect("a local write must produce a decay score");
    assert!(
        score > 0.0,
        "local write must accumulate a positive score, got {score}"
    );
}