frame-conv 0.2.0

//! F-3a R5 — backpressure: Defer ASSERTED, not assumed. The characterized
//! S4 reality row at published liminal 0.3.0 (F-0c-FINDINGS §R5): no
//! Defer/Reject-class value exists in the admission's closed answer set and
//! a slow consumer surfaces NOTHING to anyone. frame-conv adds NO parallel
//! vocabulary (that would fork the substrate's typed home before it
//! exists); when the upstream Accept/Defer/Reject wiring lands on the
//! participant surface, the pattern inherits it through the substrate enum
//! it already consumes. This test pins the current truth.

#![allow(clippy::expect_used, clippy::unwrap_used)]

mod support;

use std::error::Error;
use std::time::{Duration, Instant};

use frame_conv::{ConversationHandle, CursorCommit, SubscriptionItem};
use serde::{Deserialize, Serialize};
use support::{FileStore, QUANTUM, RunningServer, attachment, store_dir};

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct ContactAdded {
    name: String,
    total: u32,
}

/// The S4 reality row at the pattern level: every publish against a
/// non-reading subscriber commits with no pressure signal anywhere; the
/// slow consumer's eventual drain is complete, ordered, by content; its
/// post-drain cursor commit succeeds.
#[test]
fn slow_subscriber_surfaces_nothing_and_drain_is_complete() -> Result<(), Box<dyn Error>> {
    let server = RunningServer::start()?;
    let stores = store_dir("backpressure")?;

    let (mut publisher, _pub_grant) = ConversationHandle::open(
        &attachment(server.endpoint()),
        FileStore::new(stores.join("publisher.lpcr")),
    )?;
    let conversation = publisher.conversation();

    // The slow consumer: enrolled, then never reads while 12 events commit.
    let (mut subscriber, _sub_grant) = ConversationHandle::join(
        &attachment(server.endpoint()),
        conversation,
        FileStore::new(stores.join("subscriber.lpcr")),
    )?;

    let mut receipts = Vec::new();
    for total in 1..=12_u32 {
        let receipt = publisher.publish_event(&ContactAdded {
            name: format!("contact-{total}"),
            total,
        })?;
        receipts.push(receipt.seq);
    }

    // Every admission committed; the publisher observed no pressure signal
    // of any kind (none exists — the current truth, asserted not assumed).
    assert_eq!(receipts.len(), 12);
    assert!(
        receipts
            .windows(2)
            .all(|pair| pair[1].value() == pair[0].value() + 1),
        "all 12 commits must occupy contiguous sequences: {receipts:?}"
    );
    let publisher_counters = publisher.anomaly_counters();
    assert_eq!(publisher_counters.unexpected_frames, 0);
    assert_eq!(publisher_counters.gaps, 0);

    // The slow consumer's drain: complete, ordered, by content.
    let mut drained = Vec::new();
    let until = Instant::now() + 4 * QUANTUM;
    while drained.len() < 12 && Instant::now() < until {
        if let Some(SubscriptionItem::Event { body, seq, .. }) =
            subscriber.next_event::<ContactAdded>(Duration::from_secs(1))?
        {
            drained.push((body.total, seq));
        }
    }
    let expected: Vec<(u32, frame_conv::ConversationSeq)> = receipts
        .iter()
        .enumerate()
        .map(|(index, seq)| (u32::try_from(index).unwrap_or(0) + 1, *seq))
        .collect();
    assert_eq!(
        drained, expected,
        "the drain must be complete, ordered, at the committed sequences"
    );
    assert_eq!(subscriber.anomaly_counters().gaps, 0);

    let last = expected[11].1;
    let commit = subscriber.commit_cursor(last)?;
    assert!(
        matches!(commit, CursorCommit::Committed { .. }),
        "the post-drain ack must commit: {commit:?}"
    );

    std::fs::remove_dir_all(&stores)?;
    server.shutdown()?;
    Ok(())
}