iris-chat-protocol 0.3.10

#[test]
fn replayed_authorization_does_not_copy_unrelated_pending_backlog() {
    let mut fixture = remote_send_fixture();
    let events = [1, 0].map(|created| signed_app_keys(&fixture.peer_owner, &[fixture.ready_device.public_key(), fixture.late_device.public_key()], created));
    let small = events.iter().map(|event| {
        let (result, bytes) = remote_roster_tests::allocated_bytes(|| fixture.sender.ingest_app_keys_event(event));
        assert!(result.unwrap().is_empty());
        bytes
    }).collect::<Vec<_>>();
    fixture.sender.pending_group_fanouts = (0..4096).map(|index|
        pending_local_group_fanout("unrelated", Some("event"), format!("payload-{index}").as_bytes(), 5)
    ).collect();
    let before = fixture.sender.pending_group_fanouts.clone();
    for (event, small) in events.iter().zip(small) {
        let (result, large) = remote_roster_tests::allocated_bytes(|| fixture.sender.ingest_app_keys_event(event));
        eprintln!("authorization_replay small_allocated_bytes={small} backlog_allocated_bytes={large}");
        assert!(result.unwrap().is_empty());
        assert_eq!(fixture.sender.pending_group_fanouts, before);
        assert!(large <= small + 128 * 1024, "replayed authorization copied unrelated backlog: small={small}, large={large}");
    }
}

#[test]
fn replayed_local_authorization_does_not_persist_or_retry() {
    let mut fixture = remote_send_fixture();
    let event = signed_app_keys(&fixture.owner, &[fixture.device.public_key()], unix_now().get() + 1);
    fixture.sender.ingest_app_keys_event(&event).unwrap();
    let storage = Arc::new(GroupRetryCountingStorage { inner: fixture.sender.storage.clone(), puts: 0.into(), fail_next: false.into() });
    fixture.sender.storage = storage.clone();
    let before = fixture.sender.session_manager.snapshot();
    assert!(fixture.sender.ingest_app_keys_event(&event).unwrap().is_empty());
    assert!(fixture.sender.ingest_app_keys_snapshot(event.pubkey, AppKeys::from_event(&event).unwrap(), event.created_at.as_secs()).unwrap().is_empty());
    assert_eq!(fixture.sender.session_manager.snapshot(), before);
    assert_eq!(storage.puts.load(std::sync::atomic::Ordering::Relaxed), 0, "identical roster must not cause a checkpoint");
}

#[test]
fn authorization_admission_failure_restores_exact_previous_generation() {
    for local in [false, true] {
        for missing_evidence in [false, true] {
            let mut fixture = checkpoint_backlog_fixture(32, 8);
            let signer = if local { &fixture.owner } else { &fixture.peer_owner };
            let owner = ndr_owner(signer.public_key());
            if missing_evidence {
                fixture.sender.invite_owner_app_keys_evidence.remove(&owner);
                fixture.sender.verified_app_keys_owners.remove(&owner);
                if local { fixture.sender.local_app_keys_observed = false; }
            }
            fixture.sender.persist().unwrap();
            let before: serde_json::Value = serde_json::from_str(&fixture.store.get(PROTOCOL_ENGINE_STATE_KEY).unwrap().unwrap()).unwrap();
            let storage = Arc::new(GroupRetryCountingStorage { inner: fixture.store.clone(), puts: 0.into(), fail_next: true.into() });
            fixture.sender.storage = storage;
            let event = signed_app_keys(signer, &[], unix_now().get() + 1);
            assert!(fixture.sender.ingest_app_keys_event(&event).is_err());
            fixture.sender.persist().unwrap();
            let after: serde_json::Value = serde_json::from_str(&fixture.store.get(PROTOCOL_ENGINE_STATE_KEY).unwrap().unwrap()).unwrap();
            assert_eq!(after, before, "admission rollback local={local} missing_evidence={missing_evidence}");
        }
    }
}

#[test]
fn equal_timestamp_conflicting_authorization_remains_ambiguous() {
    let mut fixture = remote_send_fixture();
    let created = unix_now().get();
    let first = signed_app_keys(&fixture.peer_owner, &[fixture.ready_device.public_key()], created);
    let conflicting = signed_app_keys(&fixture.peer_owner, &[fixture.late_device.public_key()], created);
    fixture.sender.ingest_app_keys_event(&first).unwrap();
    fixture.sender.ingest_app_keys_event(&conflicting).unwrap();
    assert!(matches!(fixture.sender.invite_owner_app_keys_evidence.get(&ndr_owner(fixture.peer_owner.public_key())), Some(ProtocolAppKeysEvidence::Ambiguous { created_at_secs }) if *created_at_secs == created));
    fixture.sender.ingest_app_keys_event(&first).unwrap();
    assert!(matches!(fixture.sender.invite_owner_app_keys_evidence.get(&ndr_owner(fixture.peer_owner.public_key())), Some(ProtocolAppKeysEvidence::Ambiguous { .. })));
}

#[test]
fn checkpoint_cold_layout_recovers_core_without_matching_quoted_or_reordered_fields() {
    let prefix = format!("{{\"version\":{},\"pending_group_sender_key_messages\":[],\"pending_group_fanouts\":[],\"pending_inbound\":[]", PROTOCOL_ENGINE_STATE_VERSION);
    let sessions = serde_json::json!({"users": (0..40).map(|i|
        serde_json::json!({"i":i,"state":"x".repeat(2300)})).collect::<Vec<_>>()});
    let core = format!("{prefix},\"session_manager\":{sessions}}}");
    let (core, layout) = layout_protocol_checkpoint(core, &ProtocolCheckpointLayout::default());
    let quoted = serde_json::to_string("\"pending_inbound\":[],\"session_manager\":{} 🦊").unwrap();
    let json = format!("{{\"version\":{},\"pending_group_sender_key_messages\":[{{\"text\":{quoted}}}],\"pending_group_fanouts\":[],\"pending_inbound\":[]{}   ", PROTOCOL_ENGINE_STATE_VERSION, &core[prefix.len()..]);
    let restored = ProtocolCheckpointLayout::from_json(&json);
    let restored_core = restored.core.unwrap();
    assert_eq!(restored_core.record_ends, layout.record_ends);
    assert_eq!(restored_core.bytes, core.trim_end().len());
    // Old/reordered JSON is still readable, but its unrelated whitespace must
    // not be mistaken for this serializer's separately laid-out core.
    let reordered = format!("{{\"session_manager\":{sessions},\"version\":1,\"pending_group_sender_key_messages\":[],\"pending_group_fanouts\":[],\"pending_inbound\":[]}}");
    assert!(ProtocolCheckpointLayout::from_json(&reordered)
        .core
        .is_none());
    let nested =
        serde_json::json!({"nested": serde_json::from_str::<serde_json::Value>(&json).unwrap()})
            .to_string();
    assert!(ProtocolCheckpointLayout::from_json(&nested).core.is_none());
}

#[test]
fn checkpoint_cold_reopen_retains_core_boundaries_after_real_expiry() {
    let (mut fixture, directory, connection, storage) = checkpoint_growth_fixture();
    let peer = fixture.peer_owner.public_key();
    fixture.sender.enter_batch();
    for index in 0..33 {
        let text = format!(
            "cold-start-{index}:{}",
            "x".repeat(if index == 0 { 8_000 } else { 1_200 })
        );
        let sent = fixture
            .sender
            .send_direct_text(
                peer,
                &peer.to_hex(),
                &text,
                (index == 0).then_some(200),
                UnixSeconds(100),
            )
            .unwrap();
        assert_eq!(
            decrypt_own_sync_effects(&mut fixture.ready, sent.effects).len(),
            1
        );
    }
    fixture.sender.exit_batch().unwrap();
    let before = fixture.sender.pending_remote_sends.len();
    assert!(fixture
        .sender
        .retry_pending_remote_sends(NdrUnixSeconds(200))
        .unwrap()
        .is_empty());
    assert_eq!(fixture.sender.pending_remote_sends.len(), before - 1);
    // Real expiry leaves the following smaller records in the saved larger
    // slots. A cold load must retain those boundaries instead of repacking.
    assert!(
        fixture
            .sender
            .checkpoint_layout
            .borrow()
            .core
            .as_ref()
            .unwrap()
            .bytes
            > 128 * 1024
    );
    let saved = storage.get(PROTOCOL_ENGINE_STATE_KEY).unwrap().unwrap();
    let expected: serde_json::Value = serde_json::from_str(&saved).unwrap();
    let owner = fixture.owner.public_key();
    let device = fixture.device.clone();
    drop(saved);
    drop(fixture);
    drop(storage);
    drop(connection);

    let mut writes = Vec::new();
    for _ in 0..2 {
        let conn = rusqlite::Connection::open(directory.path().join("state.sqlite3")).unwrap();
        conn.execute_batch("PRAGMA wal_autocheckpoint=0;
            CREATE TEMP TABLE startup_payload_writes(immutable INTEGER NOT NULL, bytes INTEGER NOT NULL);
            CREATE TEMP TRIGGER startup_payload_insert AFTER INSERT ON main.ndr_kv_checkpoint_chunks BEGIN
                INSERT INTO startup_payload_writes VALUES(substr(NEW.payload_key,1,1)=x'01',length(NEW.data));
            END;
            CREATE TEMP TRIGGER startup_payload_update AFTER UPDATE ON main.ndr_kv_checkpoint_chunks BEGIN
                INSERT INTO startup_payload_writes VALUES(substr(NEW.payload_key,1,1)=x'01',length(NEW.data));
            END;").unwrap();
        let connection = Arc::new(Mutex::new(conn));
        let storage = Arc::new(SqliteStorageAdapter::new(
            connection.clone(),
            owner.to_hex(),
            device.public_key().to_hex(),
        ));
        let before = checkpoint_growth_written_bytes(&connection);
        let restored =
            ProtocolEngine::load_or_create_for_local_device(storage.clone(), owner, &device)
                .unwrap();
        let written = checkpoint_growth_written_bytes(&connection) - before;
        let (immutable, inline): (u64,u64) = connection.lock().unwrap().query_row(
            "SELECT coalesce(sum(CASE WHEN immutable THEN bytes ELSE 0 END),0),coalesce(sum(CASE WHEN NOT immutable THEN bytes ELSE 0 END),0) FROM startup_payload_writes",[],|row|Ok((row.get(0)?,row.get(1)?))
        ).unwrap();
        assert_eq!(
            immutable, 0,
            "cold caches must not reinsert immutable bodies"
        );
        let saved = storage.get(PROTOCOL_ENGINE_STATE_KEY).unwrap().unwrap();
        assert_eq!(saved.len(), restored.checkpoint_layout.borrow().bytes);
        assert!(serde_json::from_str::<serde_json::Value>(&saved).unwrap() == expected,
            "startup must preserve the complete queued state, ratchets, journal and authorization evidence");
        eprintln!("checkpoint_cold_reopen physical_write_bytes={written} immutable_insert_update_bytes={immutable} inline_insert_update_bytes={inline}");
        writes.push(written);
        drop(restored);
        drop(storage);
        drop(connection);
    }
    // The materialized JSON does not distinguish core trailing whitespace
    // from the final reserve. A first restore may normalize its last chunk.
    assert!(
        writes[0] <= 128 * 1024,
        "cold restore repacked the mutable core: {}",
        writes[0]
    );
    assert_eq!(
        writes[1], 0,
        "a second unchanged cold restore must not write data"
    );
}

#[test]
fn checkpoint_cold_reopen_migrates_legacy_once_without_losing_state() {
    let (fixture, directory, connection, storage) = checkpoint_growth_fixture();
    let saved = storage.get(PROTOCOL_ENGINE_STATE_KEY).unwrap().unwrap();
    let expected: serde_json::Value = serde_json::from_str(&saved).unwrap();
    let owner = fixture.owner.public_key();
    let device = fixture.device.clone();
    // The ordinary adapter path represents an older whole-value checkpoint;
    // no logical-record metadata survives this replacement.
    storage.put(PROTOCOL_ENGINE_STATE_KEY, saved).unwrap();
    let count: usize = connection
        .lock()
        .unwrap()
        .query_row("SELECT count(*) FROM ndr_kv_checkpoint", [], |row| {
            row.get(0)
        })
        .unwrap();
    assert_eq!(count, 0);
    drop(fixture);
    drop(storage);
    drop(connection);
    let mut writes = Vec::new();
    for _ in 0..2 {
        let connection = Arc::new(Mutex::new(
            rusqlite::Connection::open(directory.path().join("state.sqlite3")).unwrap(),
        ));
        let storage = Arc::new(SqliteStorageAdapter::new(
            connection.clone(),
            owner.to_hex(),
            device.public_key().to_hex(),
        ));
        let before = checkpoint_growth_written_bytes(&connection);
        let restored =
            ProtocolEngine::load_or_create_for_local_device(storage.clone(), owner, &device)
                .unwrap();
        writes.push(checkpoint_growth_written_bytes(&connection) - before);
        let saved = storage.get(PROTOCOL_ENGINE_STATE_KEY).unwrap().unwrap();
        assert!(serde_json::from_str::<serde_json::Value>(&saved).unwrap() == expected);
        let count: usize = connection
            .lock()
            .unwrap()
            .query_row("SELECT count(*) FROM ndr_kv_checkpoint", [], |row| {
                row.get(0)
            })
            .unwrap();
        assert_eq!(count, 1);
        drop(restored);
        drop(storage);
        drop(connection);
    }
    assert!(
        writes[0] > 0,
        "first migration must actually store the payloads"
    );
    assert_eq!(
        writes[1], 0,
        "completed migration must not recur on cold reopen"
    );
}