uqa-storage 0.5.3

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

//! Sequence catalog operations share the caller's chosen session and commit boundary.

use std::sync::Arc;

use crate::{
    CatalogFacade, KeyValueCatalog, KeyValueStore, RelationIdentity, SequenceLogResult,
    SequenceOptions, SequenceReservationResult, SequenceRow, SequenceValuePosition,
    SequenceValueReservation, StorageBackendResult,
};

fn sequence(name: &str, id: u8) -> SequenceRow {
    SequenceRow {
        relation: RelationIdentity::new("public", name),
        security: crate::SequenceSecurityRow::Bound(
            uqa_core::catalog_sequence::BoundSequenceSecurity::owner(
                uqa_core::catalog_role::RoleIdentity {
                    oid: 20_001,
                    object_id: [9; 16],
                },
            ),
        ),
        object_id: [id; 16],
        definition_generation: [id + 64; 16],
        start: 1,
        increment: 1,
        current: 1,
        called: false,
        log_count: 0,
        persistence: "p".into(),
        owner: None,
        options: SequenceOptions {
            min_value: Some(1),
            max_value: Some(i64::MAX),
            cache_size: 3,
            ..SequenceOptions::default()
        },
    }
}

fn reserve(
    catalog: &KeyValueCatalog,
    row: &SequenceRow,
) -> StorageBackendResult<SequenceValueReservation> {
    let SequenceReservationResult::Reserved(reservation) = catalog.reserve_sequence_values(
        &row.relation.qualified_name(),
        row.object_id,
        row.definition_generation,
    )?
    else {
        panic!("expected a sequence reservation");
    };
    Ok(reservation)
}

/// Exercise independent sequence consumers, selected autonomous sessions, definition conflicts and undo; leave `public.sequence_saved` for reopen verification. Use fresh disposable storage.
pub fn verify_sequence_concurrency(
    a: &Arc<dyn KeyValueStore>,
    b: &Arc<dyn KeyValueStore>,
) -> StorageBackendResult<()> {
    let first = KeyValueCatalog::new(a.clone());
    let second = KeyValueCatalog::new(b.clone());
    first.save_schema("public")?;
    let row = sequence("sequence_a", 1);
    let independent = sequence("sequence_b", 2);
    assert!(first.create_sequence_row(&row)?);
    assert!(first.create_sequence_row(&independent)?);
    a.begin_transaction()?;
    first.save_model("sequence_private", "kept")?;
    assert_eq!(reserve(&first, &row)?.first_value, 1);
    a.savepoint("sequence_position")?;
    first.set_sequence_value(
        "sequence_a",
        row.object_id,
        row.definition_generation,
        40,
        false,
        0,
    )?;
    assert_eq!(reserve(&first, &row)?.last_value, 42);
    b.begin_transaction()?;
    assert_eq!(reserve(&second, &independent)?.last_value, 3);
    b.commit_transaction()?;
    a.rollback_to_savepoint("sequence_position")?;
    a.release_savepoint("sequence_position")?;
    a.commit_transaction()?;
    assert_eq!(reserve(&first, &row)?.first_value, 4);
    assert_eq!(reserve(&first, &independent)?.first_value, 4);

    // The catalog obeys its selected session; execution chooses the autonomous sibling for committed SQL sequences.
    a.begin_transaction()?;
    first.save_model("sequence_private", "discarded")?;
    assert_eq!(reserve(&second, &row)?.first_value, 7);
    a.rollback_transaction()?;
    assert_eq!(
        first.load_model("sequence_private")?.as_deref(),
        Some("kept")
    );
    assert_eq!(reserve(&first, &row)?.first_value, 10);
    first.drop_sequence_row("sequence_a")?;
    first.drop_sequence_row("sequence_b")?;

    verify_sequence_conflicts(a, b, &first, &second)?;
    verify_sequence_records(&first)?;

    let row = sequence("sequence_saved", 6);
    first.create_sequence_row(&row)?;
    a.begin_transaction()?;
    a.savepoint("sequence_definition")?;
    let mut changed = row.clone();
    changed.definition_generation = [100; 16];
    changed.current = 100;
    first.replace_sequence_row(&changed)?;
    assert_eq!(reserve(&first, &changed)?.first_value, 100);
    first.rename_sequence_row("sequence_saved", "sequence_temporary")?;
    a.rollback_to_savepoint("sequence_definition")?;
    a.release_savepoint("sequence_definition")?;
    assert_eq!(first.load_sequence_rows()?, std::slice::from_ref(&row));
    assert_eq!(reserve(&first, &row)?.last_value, 3);
    first.rename_sequence_row("sequence_saved", "sequence_temporary")?;
    first.rename_sequence_row("sequence_temporary", "sequence_saved")?;
    a.commit_transaction()?;
    verify_sequence_reopen(a)
}

fn verify_sequence_conflicts(
    a: &Arc<dyn KeyValueStore>,
    b: &Arc<dyn KeyValueStore>,
    first: &KeyValueCatalog,
    second: &KeyValueCatalog,
) -> StorageBackendResult<()> {
    // A reservation conflicts with every other change of the generation's value record: a value change, its replacement by a new generation, or its removal.
    for change in ["reserve", "set", "replace", "drop"] {
        for change_wins in [false, true] {
            let row = sequence("sequence_conflict", 3);
            assert!(first.create_sequence_row(&row)?);
            a.begin_transaction()?;
            b.begin_transaction()?;
            assert_eq!(reserve(first, &row)?.first_value, 1);
            match change {
                "reserve" => assert_eq!(reserve(second, &row)?.first_value, 1),
                "set" => assert_eq!(
                    second.set_sequence_value(
                        "sequence_conflict",
                        row.object_id,
                        row.definition_generation,
                        100,
                        false,
                        0
                    )?,
                    crate::SequenceSetValueResult::Set(100)
                ),
                "replace" => {
                    let mut changed = row.clone();
                    changed.definition_generation = [99; 16];
                    changed.current = 100;
                    assert!(second.replace_sequence_row(&changed)?);
                }
                _ => assert!(second.drop_sequence_row("sequence_conflict")?),
            }
            let (winner, loser, catalog) = if change_wins {
                (b, a, second)
            } else {
                (a, b, first)
            };
            winner.commit_transaction()?;
            let expected = catalog.load_sequence_rows()?;
            let error = loser
                .commit_transaction()
                .expect_err("overlapping sequence writes must conflict");
            let crate::StorageBackendError::Backend { source, .. } = error else {
                panic!("expected a sequence write conflict: {error}");
            };
            assert!(
                matches!(
                    source.downcast_ref::<crate::mvcc::VersionError>(),
                    Some(crate::mvcc::VersionError::WriteConflict { .. })
                ),
                "{source}"
            );
            loser.rollback_transaction()?;
            assert_eq!(first.load_sequence_rows()?, expected);
            assert_eq!(second.load_sequence_rows()?, expected);
            first.drop_sequence_row("sequence_conflict")?;
        }
    }

    verify_definition_changes_leave_value_records(a, b, first, second)?;

    for other_wins in [false, true] {
        let first_row = sequence("sequence_claim", 4);
        let second_row = sequence("sequence_claim", 5);
        a.begin_transaction()?;
        b.begin_transaction()?;
        assert!(first.create_sequence_row(&first_row)?);
        assert!(second.create_sequence_row(&second_row)?);
        let (winner, loser, expected) = if other_wins {
            (b, a, &second_row)
        } else {
            (a, b, &first_row)
        };
        winner.commit_transaction()?;
        assert!(loser.commit_transaction().is_err());
        loser.rollback_transaction()?;
        assert_eq!(first.load_sequence_rows()?, std::slice::from_ref(expected));
        first.drop_sequence_row("sequence_claim")?;
    }

    Ok(())
}

/// A change of the definition alone leaves the value record, as `PostgreSQL`'s catalog rows hold a sequence's name and privileges apart from its state, so a reservation and such a change both commit.
fn verify_definition_changes_leave_value_records(
    a: &Arc<dyn KeyValueStore>,
    b: &Arc<dyn KeyValueStore>,
    first: &KeyValueCatalog,
    second: &KeyValueCatalog,
) -> StorageBackendResult<()> {
    for change in ["rename", "owner"] {
        let row = sequence("sequence_definition", 3);
        assert!(first.create_sequence_row(&row)?);
        a.begin_transaction()?;
        b.begin_transaction()?;
        assert_eq!(reserve(first, &row)?.first_value, 1);
        let mut expected = row.clone();
        if change == "rename" {
            assert!(second.rename_sequence_row("sequence_definition", "sequence_renamed")?);
            expected.relation = RelationIdentity::new("public", "sequence_renamed");
        } else {
            expected.security = crate::SequenceSecurityRow::Bound(
                uqa_core::catalog_sequence::BoundSequenceSecurity::owner(
                    uqa_core::catalog_role::RoleIdentity {
                        oid: 20_002,
                        object_id: [8; 16],
                    },
                ),
            );
            assert!(second.replace_sequence_row(&expected)?);
        }
        a.commit_transaction()?;
        b.commit_transaction()?;
        expected.current = 3;
        expected.called = true;
        expected.log_count = 32;
        assert_eq!(first.load_sequence_rows()?, [expected.clone()], "{change}");
        assert_eq!(second.load_sequence_rows()?, [expected.clone()], "{change}");
        assert!(first.drop_sequence_row(&expected.relation.qualified_name())?);
    }

    Ok(())
}

/// A record moves ahead only from the position its caller read, and a replacement that keeps the allocation generation keeps the record's value state.
fn verify_sequence_records(catalog: &KeyValueCatalog) -> StorageBackendResult<()> {
    let row = sequence("sequence_record", 8);
    assert!(catalog.create_sequence_row(&row)?);
    let name = row.relation.qualified_name();
    let logged = |value| SequenceValuePosition {
        current: value,
        called: true,
        log_count: 0,
    };
    let log = |expected, value| {
        catalog.log_sequence_values(
            &name,
            row.object_id,
            row.definition_generation,
            expected,
            logged(value),
        )
    };
    let stored = || -> StorageBackendResult<SequenceRow> {
        Ok(catalog
            .load_sequence_rows()?
            .into_iter()
            .find(|stored| stored.relation == row.relation)
            .expect("the sequence row"))
    };
    assert_eq!(log((1, false), 33)?, SequenceLogResult::Logged);
    // The record is no longer the one this caller read: nothing is written, and the record it holds is reported.
    assert_eq!(log((1, false), 99)?, SequenceLogResult::Changed(logged(33)));
    assert_eq!(
        log((33, false), 99)?,
        SequenceLogResult::Changed(logged(33))
    );
    assert_eq!(log((33, true), 66)?, SequenceLogResult::Logged);
    assert_eq!(
        catalog.log_sequence_values(&name, row.object_id, [200; 16], (66, true), logged(99))?,
        SequenceLogResult::DefinitionChanged
    );
    assert_eq!(
        catalog.log_sequence_values(
            &name,
            [201; 16],
            row.definition_generation,
            (66, true),
            logged(99)
        )?,
        SequenceLogResult::Missing
    );
    // The value record is found by object identity and generation, whatever name the caller knows the sequence by, so that an operation outside a transaction that renames the sequence finds it.
    assert_eq!(
        catalog.log_sequence_values(
            "public.sequence_absent",
            row.object_id,
            row.definition_generation,
            (66, true),
            logged(66)
        )?,
        SequenceLogResult::Logged
    );
    assert!(catalog
        .log_sequence_values(
            &name,
            row.object_id,
            row.definition_generation,
            (66, true),
            SequenceValuePosition {
                log_count: -1,
                ..logged(99)
            },
        )
        .is_err());
    let record = stored()?;
    assert_eq!(
        (record.current, record.called, record.log_count),
        (66, true, 0)
    );

    // A replacement of the same allocation generation changes the definition and leaves the value state, which a session's registry does not hold.
    let mut replaced = row.clone();
    replaced.start = 7;
    replaced.current = 5;
    replaced.called = false;
    replaced.log_count = 9;
    assert!(catalog.replace_sequence_row(&replaced)?);
    let record = stored()?;
    assert_eq!(
        (
            record.start,
            record.current,
            record.called,
            record.log_count
        ),
        (7, 66, true, 0)
    );
    // A new allocation generation is stored with the value state it comes with.
    replaced.definition_generation = [202; 16];
    assert!(catalog.replace_sequence_row(&replaced)?);
    assert_eq!(stored()?, replaced);
    assert!(catalog.drop_sequence_row(&name)?);
    Ok(())
}

/// Verify the committed position, object identity and released rename destination left by `verify_sequence_concurrency` after closing all original handles.
pub fn verify_sequence_reopen(store: &Arc<dyn KeyValueStore>) -> StorageBackendResult<()> {
    let catalog = KeyValueCatalog::new(store.clone());
    let mut expected = sequence("sequence_saved", 6);
    expected.current = 3;
    expected.called = true;
    expected.log_count = 32;
    assert_eq!(catalog.load_sequence_rows()?, [expected]);
    let probe = sequence("sequence_temporary", 7);
    assert!(catalog.create_sequence_row(&probe)?);
    assert!(catalog.drop_sequence_row("sequence_temporary")?);
    Ok(())
}