pipecircuit 0.0.19

use super::*;
use crate::storage::warp_table::observed::*;
use alloc::{format, string::String};
use core::num::NonZeroU64;
use proptest::prelude::*;

const REQUEST: RequestId = RequestId(NonZeroU64::MIN);
const NEXT: RequestId = RequestId(NonZeroU64::MIN.saturating_add(1));
type Request = ContinuousRequest<(), u32, ()>;
type Reply = ContinuousReply<(), u32, u8, u32, u32>;
type CommandReply = Result<TableReply<(), u32, u8>, StorageFailure>;
fn debug(error: impl core::fmt::Debug) -> String {
    format!("{error:?}")
}
fn receipt(sequence: u64) -> Receipt {
    Receipt::from_operation(OperationId {
        history: HistoryId([7; 16]),
        epoch: 1,
        session: SessionId(1),
        sequence,
    })
}
fn command(guarantee: Guarantee) -> Result<PendingMutation<u32>, String> {
    PendingMutation::new(TableRequest::<(), _>::Mutate {
        receipt: receipt(1),
        command: 5,
        guarantee,
    })
    .map_err(debug)
}
fn confirmed(id: Receipt, guarantee: Guarantee) -> CommandReply {
    Ok(TableReply::Mutation(MutationAnswer::Confirmed {
        receipt: id,
        outcome: Ok(15),
        established: guarantee,
    }))
}

#[test]
fn exact_operation_survives_disconnect_unknown_and_explicit_retry() -> Result<(), String> {
    let mut pending = command(Guarantee::LocallyDurable)?;
    assert_eq!(pending.status(), MutationStatus::Ready);
    assert_eq!(pending.command(), &5);
    assert_eq!(pending.guarantee(), Guarantee::LocallyDurable);
    assert!(matches!(
        pending.inquire::<(), ()>(REQUEST),
        Err(MutationStatus::Ready)
    ));
    assert_eq!(pending.disconnect(), MutationStatus::Ready);
    let first: Request = pending.retry(REQUEST).map_err(debug)?;
    assert!(
        matches!(first, Request::Command { request: TableRequest::Mutate { receipt: r, command: 5, guarantee: Guarantee::LocallyDurable }, .. } if r == receipt(1))
    );
    assert!(matches!(
        pending.retry::<(), ()>(NEXT),
        Err(MutationStatus::AwaitingReply)
    ));
    assert!(matches!(
        pending.inquire::<(), ()>(NEXT),
        Err(MutationStatus::AwaitingReply)
    ));
    assert_eq!(pending.disconnect(), MutationStatus::Uncertain);
    assert_eq!(
        pending.step(ReceiveMutation {
            id: REQUEST,
            reply: &confirmed(receipt(1), Guarantee::LocallyDurable)
        }),
        Err(MutationReceiveError::Unexpected)
    );
    let inquiry: Request = pending.inquire(NEXT).map_err(debug)?;
    assert!(
        matches!(inquiry, Request::Command { id: NEXT, request: TableRequest::Inquire { receipt: r, guarantee: Guarantee::LocallyDurable } } if r == receipt(1))
    );
    let unknown: CommandReply = Ok(TableReply::Mutation(MutationAnswer::Unknown));
    assert_eq!(
        pending.step(ReceiveMutation {
            id: NEXT,
            reply: &unknown
        }),
        Ok(MutationStatus::Uncertain)
    );
    let retry: Request = pending.retry(REQUEST).map_err(debug)?;
    assert!(
        matches!(retry, Request::Command { request: TableRequest::Mutate { receipt: r, command: 5, .. }, .. } if r == receipt(1))
    );
    let progress: CommandReply = Ok(TableReply::Mutation(MutationAnswer::Pending {
        receipt: receipt(1),
        outcome: Ok(15),
        established: Guarantee::Accepted,
    }));
    assert_eq!(
        pending.step(ReceiveMutation {
            id: REQUEST,
            reply: &progress
        }),
        Ok(MutationStatus::AwaitingReply)
    );
    assert_eq!(
        pending.step(ReceiveMutation {
            id: REQUEST,
            reply: &confirmed(receipt(1), Guarantee::ReplicatedDurable)
        }),
        Ok(MutationStatus::Confirmed)
    );
    assert_eq!(pending.disconnect(), MutationStatus::Confirmed);
    assert!(matches!(
        pending.retry::<(), ()>(NEXT),
        Err(MutationStatus::Confirmed)
    ));
    assert!(matches!(
        pending.inquire::<(), ()>(NEXT),
        Err(MutationStatus::Confirmed)
    ));
    assert!(
        matches!(pending.into_request::<()>(), TableRequest::Mutate { receipt: r, command: 5, guarantee: Guarantee::LocallyDurable } if r == receipt(1))
    );
    Ok(())
}

#[test]
fn mutation_failures_and_wrong_evidence_preserve_owned_command() -> Result<(), String> {
    for failure in [
        Ok(TableReply::Mutation(MutationAnswer::Unknown)),
        Ok(TableReply::Mutation(MutationAnswer::Unavailable(
            Error::ExpiredOperation,
        ))),
        Err(StorageFailure::WorkerClosed),
    ] {
        let mut pending = command(Guarantee::LocallyDurable)?;
        let _: Request = pending.retry(REQUEST).map_err(debug)?;
        let refused: CommandReply = failure;
        assert_eq!(
            pending.step(ReceiveMutation {
                id: REQUEST,
                reply: &refused
            }),
            Ok(MutationStatus::Uncertain)
        );
        assert_eq!(pending.receipt(), receipt(1));
        assert_eq!(pending.command(), &5);
    }
    for request in [
        TableRequest::<(), u32>::Read {
            query: (),
            guarantee: Guarantee::Accepted,
        },
        TableRequest::Inquire {
            receipt: receipt(1),
            guarantee: Guarantee::LocallyDurable,
        },
        TableRequest::Retire {
            receipt: receipt(1),
        },
    ] {
        let before = debug(&request);
        let returned = match PendingMutation::new(request) {
            Err(returned) => returned,
            Ok(_) => return Err("non-mutation accepted".into()),
        };
        assert_eq!(debug(returned), before);
    }
    for required in [
        Guarantee::Accepted,
        Guarantee::LocallyDurable,
        Guarantee::ReplicatedDurable,
    ] {
        for established in [
            Guarantee::Accepted,
            Guarantee::LocallyDurable,
            Guarantee::ReplicatedDurable,
        ] {
            let mut pending = command(required)?;
            let _: Request = pending.retry(REQUEST).map_err(debug)?;
            for reply in [
                Ok(TableReply::Read(Answer::Ready(Ok(())))),
                Ok(TableReply::Retired(Ok(()))),
            ] {
                let reply: CommandReply = reply;
                assert_eq!(
                    pending.step(ReceiveMutation {
                        id: REQUEST,
                        reply: &reply
                    }),
                    Err(MutationReceiveError::Unexpected)
                );
            }
            let wrong_pending: CommandReply = Ok(TableReply::Mutation(MutationAnswer::Pending {
                receipt: receipt(2),
                outcome: Ok(15),
                established,
            }));
            assert_eq!(
                pending.step(ReceiveMutation {
                    id: REQUEST,
                    reply: &wrong_pending
                }),
                Err(MutationReceiveError::WrongReceipt)
            );
            assert_eq!(
                pending.step(ReceiveMutation {
                    id: REQUEST,
                    reply: &confirmed(receipt(2), established)
                }),
                Err(MutationReceiveError::WrongReceipt)
            );
            assert_eq!(
                pending.step(ReceiveMutation {
                    id: NEXT,
                    reply: &confirmed(receipt(1), established)
                }),
                Err(MutationReceiveError::Unexpected)
            );
            let accepted = matches!(
                (required, established),
                (Guarantee::Accepted, _)
                    | (
                        Guarantee::LocallyDurable,
                        Guarantee::LocallyDurable | Guarantee::ReplicatedDurable
                    )
                    | (Guarantee::ReplicatedDurable, Guarantee::ReplicatedDurable)
            );
            // A durable domain refusal is a known outcome too; the caller owns it.
            let refusal: CommandReply = Ok(TableReply::Mutation(MutationAnswer::Confirmed {
                receipt: receipt(1),
                outcome: Err(9),
                established,
            }));
            let result = pending.step(ReceiveMutation {
                id: REQUEST,
                reply: &refusal,
            });
            assert_eq!(
                result,
                if accepted {
                    Ok(MutationStatus::Confirmed)
                } else {
                    Err(MutationReceiveError::InsufficientGuarantee)
                }
            );
            assert_eq!(
                pending.status(),
                if accepted {
                    MutationStatus::Confirmed
                } else {
                    MutationStatus::AwaitingReply
                }
            );
        }
    }
    Ok(())
}

#[derive(Clone, Eq, PartialEq)]
struct All;
impl ViewDefinition<Stock> for All {
    type Value = u32;
    fn project(&self, row: StockRef<'_>) -> Option<u32> {
        Some(*row.quantity)
    }
}
fn subscription_ids() -> Result<(SubscriptionId, SubscriptionId), String> {
    let authority = fresh(Replication::Disabled).map_err(debug)?;
    let mut subscriptions = Subscriptions::new(SubscriptionLimits {
        views: limits().rows,
        subscribers: limits().rows,
        retained_changes: limits().rows,
    })
    .map_err(debug)?;
    let first = subscriptions
        .subscribe(&authority, All, Guarantee::Accepted)
        .map_err(debug)?;
    let second = subscriptions
        .subscribe(&authority, All, Guarantee::Accepted)
        .map_err(debug)?;
    Ok((first, second))
}
fn stamp(revision: u64) -> ViewRevision {
    ViewRevision {
        source: SourceId::new(NonZeroU64::MIN),
        view: ViewId::new(NonZeroU64::MIN),
        epoch: 1,
        revision,
    }
}
fn baseline(subscription: SubscriptionId, revision: u64, value: u32) -> Reply {
    Reply::View {
        subscription,
        update: ViewUpdate::Baseline {
            through: stamp(revision),
            rows: vec![ViewChange::Upsert {
                id: 1,
                incarnation: 1,
                value,
            }],
        },
    }
}
fn apply(
    client: &mut SubscribedView<u32, u32>,
    session: SubscriptionSession,
    reply: &Reply,
) -> Result<SubscriptionProgress, SubscriptionReceiveError> {
    client.step(ReceiveSubscription { session, reply })
}
fn start(client: &mut SubscribedView<u32, u32>) -> Result<SubscriptionSession, String> {
    let SubscriptionStart { session, request }: SubscriptionStart<(), u32, ()> = client
        .subscribe(REQUEST, (), Guarantee::LocallyDurable)
        .map_err(debug)?;
    assert!(matches!(
        request,
        Request::Subscribe {
            id: REQUEST,
            view: (),
            guarantee: Guarantee::LocallyDurable
        }
    ));
    Ok(session)
}

#[test]
fn new_session_fences_old_traffic_even_when_wire_ids_repeat() -> Result<(), String> {
    let (subscription, wrong) = subscription_ids()?;
    let mut client = SubscribedView::new(ContinuousView::new(limits().rows));
    assert_eq!(client.view().connection(), ViewConnection::Disconnected);
    assert!(matches!(
        client.resynchronize::<(), u32, ()>(NEXT),
        Err(Error::Busy)
    ));
    let first = start(&mut client)?;
    assert!(matches!(
        client.subscribe::<(), u32, _>(NEXT, (), Guarantee::Accepted),
        Err(((), Error::Busy))
    ));
    assert!(matches!(
        client.subscribe::<(), u32, _>(NEXT, 123, Guarantee::Accepted),
        Err((123, Error::Busy))
    ));
    assert!(matches!(
        client.resynchronize::<(), u32, ()>(NEXT),
        Err(Error::Busy)
    ));
    assert!(matches!(
        apply(&mut client, first, &baseline(subscription, 1, 10)),
        Err(SubscriptionReceiveError::Unexpected)
    ));
    assert!(matches!(
        apply(
            &mut client,
            first,
            &Reply::Subscribed {
                id: NEXT,
                subscription
            }
        ),
        Err(SubscriptionReceiveError::Unexpected)
    ));
    assert_eq!(
        apply(
            &mut client,
            first,
            &Reply::Subscribed {
                id: REQUEST,
                subscription
            }
        )
        .map_err(debug)?,
        SubscriptionProgress::Subscribed
    );
    assert!(matches!(
        apply(&mut client, first, &baseline(wrong, 1, 10)),
        Err(SubscriptionReceiveError::Unexpected)
    ));
    apply(&mut client, first, &baseline(subscription, 5, 10)).map_err(debug)?;
    assert_eq!(client.view().connection(), ViewConnection::Connected);
    client.disconnect(first).map_err(debug)?;
    assert_eq!(client.view().get(1), Some(&10));
    let second = start(&mut client)?;
    assert_ne!(first, second);
    for reply in [
        Reply::Subscribed {
            id: REQUEST,
            subscription,
        },
        baseline(subscription, 100, 999),
    ] {
        assert!(matches!(
            apply(&mut client, first, &reply),
            Err(SubscriptionReceiveError::WrongSession)
        ));
    }
    assert!(matches!(
        client.disconnect(first),
        Err(SubscriptionReceiveError::WrongSession)
    ));
    apply(
        &mut client,
        second,
        &Reply::Subscribed {
            id: REQUEST,
            subscription,
        },
    )
    .map_err(debug)?;
    let mut malformed = baseline(subscription, 1, 20);
    if let Reply::View {
        update: ViewUpdate::Baseline { rows, .. },
        ..
    } = &mut malformed
    {
        rows.push(rows.first().ok_or("empty baseline")?.clone());
    }
    assert!(matches!(
        apply(&mut client, second, &malformed),
        Err(SubscriptionReceiveError::Apply(_))
    ));
    assert_eq!(client.view().get(1), Some(&10));
    assert_eq!(client.view().connection(), ViewConnection::Disconnected);
    // Fresh sessions may have a wholly new view identity/revision sequence.
    apply(&mut client, second, &baseline(subscription, 1, 20)).map_err(debug)?;
    assert_eq!(client.view().get(1), Some(&20));
    let copy = client.into_view();
    assert_eq!(copy.get(1), Some(&20));
    assert_eq!(copy.connection(), ViewConnection::Disconnected);
    Ok(())
}

#[test]
fn gaps_resynchronize_without_regression_or_unrelated_identity_replacement() -> Result<(), String> {
    let (subscription, wrong) = subscription_ids()?;
    let mut client = SubscribedView::new(ContinuousView::new(limits().rows));
    let session = start(&mut client)?;
    apply(
        &mut client,
        session,
        &Reply::Subscribed {
            id: REQUEST,
            subscription,
        },
    )
    .map_err(debug)?;
    let gap = Reply::View {
        subscription,
        update: ViewUpdate::Changes {
            base: stamp(2),
            through: stamp(3),
            changes: vec![ViewChange::Upsert {
                id: 1,
                incarnation: 1,
                value: 30,
            }],
        },
    };
    assert_eq!(
        apply(&mut client, session, &gap).map_err(debug)?,
        SubscriptionProgress::Updated(ViewApplyOutcome::BaselineRequired)
    );
    let _: Request = client.resynchronize(NEXT).map_err(debug)?;
    assert!(matches!(
        client.resynchronize::<(), u32, ()>(NEXT),
        Err(Error::Busy)
    ));
    assert!(matches!(
        apply(&mut client, session, &baseline(subscription, 1, 10)),
        Err(SubscriptionReceiveError::Unexpected)
    ));
    apply(
        &mut client,
        session,
        &Reply::Resynchronizing {
            id: NEXT,
            subscription,
        },
    )
    .map_err(debug)?;
    apply(&mut client, session, &baseline(subscription, 1, 10)).map_err(debug)?;
    assert_eq!(
        apply(&mut client, session, &gap).map_err(debug)?,
        SubscriptionProgress::Updated(ViewApplyOutcome::BaselineRequired)
    );
    assert_eq!(client.view().get(1), Some(&10));
    let _: Request = client.resynchronize(NEXT).map_err(debug)?;
    assert!(matches!(
        apply(
            &mut client,
            session,
            &Reply::Resynchronizing {
                id: REQUEST,
                subscription
            }
        ),
        Err(SubscriptionReceiveError::Unexpected)
    ));
    assert!(matches!(
        apply(
            &mut client,
            session,
            &Reply::Resynchronizing {
                id: NEXT,
                subscription: wrong
            }
        ),
        Err(SubscriptionReceiveError::Unexpected)
    ));
    apply(
        &mut client,
        session,
        &Reply::Resynchronizing {
            id: NEXT,
            subscription,
        },
    )
    .map_err(debug)?;
    assert_eq!(
        apply(&mut client, session, &gap).map_err(debug)?,
        SubscriptionProgress::Updated(ViewApplyOutcome::BaselineRequired)
    );
    let mut wrong_source = baseline(subscription, 3, 30);
    if let Reply::View {
        update: ViewUpdate::Baseline { through, .. },
        ..
    } = &mut wrong_source
    {
        through.source = SourceId::new(NonZeroU64::MIN.saturating_add(1));
    }
    assert!(matches!(
        apply(&mut client, session, &wrong_source),
        Err(SubscriptionReceiveError::Apply(_))
    ));
    assert_eq!(
        apply(&mut client, session, &baseline(subscription, 1, 999)).map_err(debug)?,
        SubscriptionProgress::Updated(ViewApplyOutcome::Stale)
    );
    assert_eq!(client.view().connection(), ViewConnection::Disconnected);
    assert_eq!(client.view().get(1), Some(&10));
    // A refused baseline does not wedge recovery; another explicit request
    // still uses this subscription's original identity and revision floor.
    let _: Request = client.resynchronize(NEXT).map_err(debug)?;
    apply(
        &mut client,
        session,
        &Reply::Resynchronizing {
            id: NEXT,
            subscription,
        },
    )
    .map_err(debug)?;
    apply(&mut client, session, &baseline(subscription, 3, 30)).map_err(debug)?;
    assert_eq!(client.view().get(1), Some(&30));
    assert_eq!(client.view().connection(), ViewConnection::Connected);
    assert_eq!(
        apply(&mut client, session, &gap).map_err(debug)?,
        SubscriptionProgress::Updated(ViewApplyOutcome::Stale)
    );
    Ok(())
}

#[test]
fn refused_subscriptions_preserve_copy_and_reject_unrelated_messages() -> Result<(), String> {
    let (subscription, _) = subscription_ids()?;
    let mut client = SubscribedView::new(ContinuousView::new(limits().rows));
    for synchronize in [false, true] {
        let session = start(&mut client)?;
        let id = if synchronize {
            apply(
                &mut client,
                session,
                &Reply::Subscribed {
                    id: REQUEST,
                    subscription,
                },
            )
            .map_err(debug)?;
            apply(&mut client, session, &baseline(subscription, 1, 10)).map_err(debug)?;
            let _: Request = client.resynchronize(NEXT).map_err(debug)?;
            NEXT
        } else {
            REQUEST
        };
        for reply in [
            Reply::Unsubscribed { id, subscription },
            Reply::Command {
                id,
                reply: confirmed(receipt(1), Guarantee::Accepted),
            },
            Reply::ReplicationInterrupted {
                failure: StorageFailure::WorkerClosed,
            },
        ] {
            assert!(matches!(
                apply(&mut client, session, &reply),
                Err(SubscriptionReceiveError::Unexpected)
            ));
        }
        assert_eq!(
            apply(
                &mut client,
                session,
                &Reply::SubscriptionRefused {
                    id,
                    reason: SubscriptionRefusal::Full
                }
            )
            .map_err(debug)?,
            SubscriptionProgress::Refused(SubscriptionRefusal::Full)
        );
        assert_eq!(client.view().connection(), ViewConnection::Disconnected);
        assert!(matches!(
            apply(&mut client, session, &baseline(subscription, 2, 90)),
            Err(SubscriptionReceiveError::Unexpected)
        ));
    }
    assert_eq!(client.view().get(1), Some(&10));
    Ok(())
}

proptest! {
    #[test]
    fn arbitrary_delivery_interleavings_never_replace_operation_or_invent_confirmation(events in proptest::collection::vec(0u8..8, 0..128)) {
        let mut pending = command(Guarantee::LocallyDurable).map_err(TestCaseError::fail)?;
        let mut known = false;
        for event in events {
            match event {
                0 => { let _request = pending.retry::<(), ()>(REQUEST); }
                1 => { pending.disconnect(); }
                2 => { let _request = pending.inquire::<(), ()>(NEXT); }
                3..=7 => {
                    let reply = match event {
                        3 => confirmed(receipt(2), Guarantee::LocallyDurable),
                        4 => confirmed(receipt(1), Guarantee::Accepted),
                        5 => Ok(TableReply::Mutation(MutationAnswer::Unknown)),
                        6 => Err(StorageFailure::WorkerClosed),
                        7 => confirmed(receipt(1), Guarantee::LocallyDurable),
                        other => return Err(TestCaseError::fail(format!("unmodeled reply {other}"))),
                    };
                    let result = pending.step(ReceiveMutation { id: NEXT, reply: &reply });
                    if result == Ok(MutationStatus::Confirmed) { prop_assert_eq!(event, 7); known = true; }
                }
                other => return Err(TestCaseError::fail(format!("unmodeled event {other}"))),
            }
            prop_assert_eq!(pending.receipt(), receipt(1));
            prop_assert_eq!(pending.command(), &5);
            prop_assert_eq!(pending.status() == MutationStatus::Confirmed, known);
        }
    }
    #[test]
    fn arbitrary_old_session_traffic_never_changes_a_reconnected_copy(events in proptest::collection::vec(any::<u32>(), 0..64)) {
        let (subscription, _) = subscription_ids().map_err(TestCaseError::fail)?;
        let mut client = SubscribedView::new(ContinuousView::new(limits().rows));
        let old = start(&mut client).map_err(TestCaseError::fail)?;
        client.disconnect(old).map_err(|e| TestCaseError::fail(debug(e)))?;
        let current = start(&mut client).map_err(TestCaseError::fail)?;
        apply(&mut client, current, &Reply::Subscribed { id: REQUEST, subscription }).map_err(|e| TestCaseError::fail(debug(e)))?;
        apply(&mut client, current, &baseline(subscription, 1, 20)).map_err(|e| TestCaseError::fail(debug(e)))?;
        for value in events {
            prop_assert!(matches!(apply(&mut client, old, &baseline(subscription, 2, value)), Err(SubscriptionReceiveError::WrongSession)));
            prop_assert!(matches!(client.disconnect(old), Err(SubscriptionReceiveError::WrongSession)));
            prop_assert_eq!(client.view().get(1), Some(&20));
            prop_assert_eq!(client.view().connection(), ViewConnection::Connected);
        }
    }
}