haematite-cli 0.35.1

//! Lane 27: a merged list longer than one row is written in pieces of at
//! most `ROW_BYTES`, and a merge of a branch that took such a list in lists
//! it again whole, in order, with the branch's own stages after it.

use std::error::Error;

use super::{
    At, Found, Listed, Merged, ROW_BYTES, decode, decode_piece, key, key_parts, merged_rows,
    stage_write,
};
use crate::serve::protocol::{Stage, StageInput, Version};

type TestResult = Result<(), Box<dyn Error>>;

const fn at(generation: u64, sequence: u64) -> Version {
    Version {
        generation,
        sequence,
    }
}

fn stage(index: u64) -> Stage {
    Stage {
        line: "line".to_owned(),
        run: "run-1".to_owned(),
        stage: format!("s{index}"),
        subject: "ward-7".to_owned(),
        inputs: vec![StageInput {
            branch: "line".to_owned(),
            version: at(5, index - 1),
        }],
    }
}

fn listed(index: u64) -> Listed {
    Listed {
        branch: "line".to_owned(),
        version: at(5, index),
        stage: stage(index),
    }
}

/// The whole list the rows written at `version` hold, read back as history
/// reads them: the merged row, then each piece it names.
fn read_back(
    version: Version,
    writes: &[(Vec<u8>, Option<Vec<u8>>)],
) -> Result<Merged, Box<dyn Error>> {
    let value = |wanted: &[u8]| {
        writes
            .iter()
            .find(|(written, _)| written.as_slice() == wanted)
            .and_then(|(_, value)| value.clone())
            .ok_or("a row is missing")
    };
    let row = decode(version, &value(&key(version))?)?;
    let mut merged = row.merged.ok_or("no merged row")?;
    for index in 0..row.pieces.unwrap_or(0) {
        let piece = value(&super::piece_key(version, index))?;
        merged.stages.extend(decode_piece(version, index, &piece)?);
    }
    Ok(merged)
}

#[test]
fn a_merged_list_in_pieces_is_listed_again_whole_and_in_order() -> TestResult {
    let merged_at = at(7, 4_200);
    let stages: Vec<Listed> = (1..=4_100).map(listed).collect();
    let from = At {
        branch: "line".to_owned(),
        version: at(5, 4_100),
    };
    let writes = merged_rows(
        merged_at,
        Merged {
            from: from.clone(),
            stages: stages.clone(),
        },
    )?;
    assert!(
        writes.len() > 2,
        "{} rows: a row and its pieces",
        writes.len()
    );
    for (written, value) in &writes {
        let bytes = value.as_ref().ok_or("a merge deletes no stage row")?;
        assert!(bytes.len() <= ROW_BYTES, "{} bytes", bytes.len());
        assert_eq!(key_parts(written)?.0, merged_at);
    }
    let whole = read_back(merged_at, &writes)?;
    assert_eq!(whole.from, from);
    assert_eq!(whole.stages, stages, "every stage, in order");
    // A branch that took the list in, and then committed a stage of its own,
    // is merged: its change pages find the rows in any order.
    let own_at = at(7, 4_201);
    let (own_key, own_value) = stage_write(own_at, &stage(4_101))?;
    let mut found = Found::default();
    found.add("work", &own_key, &own_value.ok_or("no stage row")?)?;
    for (written, value) in writes.iter().rev() {
        found.add("work", written, value.as_ref().ok_or("no value")?)?;
    }
    let again_at = at(9, 1);
    let again = found.write(again_at, "work", own_at)?;
    let listed_again = read_back(again_at, &again)?;
    let mut expected = stages;
    expected.push(Listed {
        branch: "work".to_owned(),
        version: own_at,
        stage: stage(4_101),
    });
    assert_eq!(
        listed_again.stages, expected,
        "the list again, then its own"
    );
    Ok(())
}

#[test]
fn a_short_merged_list_is_one_row_as_before() -> TestResult {
    let merged_at = at(7, 4);
    let stages: Vec<Listed> = (1..=3).map(listed).collect();
    let writes = merged_rows(
        merged_at,
        Merged {
            from: At {
                branch: "line".to_owned(),
                version: at(5, 3),
            },
            stages: stages.clone(),
        },
    )?;
    assert_eq!(writes.len(), 1, "one row");
    let bytes = writes
        .first()
        .and_then(|(_, value)| value.as_ref())
        .ok_or("no row")?;
    let row: serde_json::Value = serde_json::from_slice(bytes)?;
    assert_eq!(row.get("pieces"), None, "{row}");
    assert_eq!(read_back(merged_at, &writes)?.stages, stages);
    Ok(())
}

#[test]
fn a_merged_list_missing_a_piece_is_refused_by_name() -> TestResult {
    let merged_at = at(7, 4_200);
    let writes = merged_rows(
        merged_at,
        Merged {
            from: At {
                branch: "line".to_owned(),
                version: at(5, 4_100),
            },
            stages: (1..=4_100).map(listed).collect(),
        },
    )?;
    let mut found = Found::default();
    for (written, value) in writes
        .iter()
        .filter(|(written, _)| key_parts(written).is_ok_and(|(_, piece)| piece != Some(1)))
    {
        found.add("work", written, value.as_ref().ok_or("no value")?)?;
    }
    let refused = found
        .write(at(9, 1), "work", merged_at)
        .err()
        .ok_or("a list missing a piece was merged")?;
    assert_eq!(refused.code, "stage_row_corrupt", "{refused}");
    Ok(())
}