Function wal_redispatch_makes_data_queryable

nodedb/tests/wal_catchup.rs:151–188

()

/// catch-up reads WAL → re-dispatches → query sees the data.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn wal_redispatch_makes_data_queryable() {
    let stack = TestStack::new();
    tokio::time::sleep(Duration::from_millis(50)).await;

    let collection = "wal_test";

    // Write 500 rows directly to WAL (bypassing Data Plane dispatch).
    stack.write_to_wal(
        collection,
        ilp_payload(collection, 500, 1_700_000_000_000_000_000),
    );

    // Read WAL records and manually re-dispatch (simulating catch-up logic).
    let records = stack.wal.replay_from(nodedb_types::Lsn::new(0)).unwrap();
    assert_eq!(records.len(), 1, "WAL should have 1 timeseries batch");

    for record in &records {
        let (coll, payload): (String, Vec<u8>) =
            zerompk::from_msgpack::<(String, Vec<u8>)>(&record.payload).unwrap();
        let resp = stack
            .dispatch(
                PhysicalPlan::Timeseries(TimeseriesOp::Ingest {
                    collection: coll,
                    payload,
                    format: "ilp".to_string(),
                    wal_lsn: Some(record.header.lsn),
                    surrogates: Vec::new(),
                }),
                collection,
            )
            .await;
        assert_eq!(resp["accepted"].as_u64().unwrap(), 500);
    }

    // Query: all 500 rows should be visible.
    let count = stack.query_count(collection).await;
    assert_eq!(count, 500, "all WAL-redispatched rows should be queryable");
}

/// The catch-up background task automatically re-dispatches WAL records.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]