win-durable-ioring 0.4.0

// Copyright (c) 2026 Mike Grier
//! The consumer's durability provider (DI-3.2.6, DI-D-27, DI-D-44), against real files: what a
//! seal asks it, its answers from inside `make_durable` and from another thread, what its failure
//! and its silence are reported as, and what a panic in it does; and the routing provider in that
//! place (DI-12.2.4), composing two providers. Every entry is reported to the conformance oracle.

use std::panic::{self, AssertUnwindSafe};
use std::sync::{Arc, Mutex, mpsc};
use std::time::{Duration, Instant};

use windows_threadpool_sys::wait::ThreadpoolWait;

use super::pushes::{BOUND, GIVEN, Harness, Ring, temp};
use super::{
    FlushDomain, FlushRequest, Keeper, Kept, TargetSetup, TempFile, V, domain, file, keeper, setup,
};
use durable_ioring::DurabilityProvider;
use durable_ioring::DurableRing;
use durable_ioring::RoutingProvider;
use durable_ioring::backend::SubmissionData;
use durable_ioring::{
    DurabilityRequest, Entry, EpochState, Failed, FailureCause, IoTargetKey, OpId, OpKind,
    WriteOptions,
};

/// `ERROR_IO_DEVICE`, the error a failing provider reports.
const IO_DEVICE: i32 = 1117;

/// An instance over `temp`, as `GIVEN` in `domains`, with `provider`.
fn instance(
    temp: &TempFile,
    domains: &[&str],
    provider: Option<Box<dyn DurabilityProvider<V>>>,
) -> Harness {
    Harness::new(
        Ring::new(setup(
            vec![file(GIVEN.0, temp, domains)],
            Vec::new(),
            provider,
        ))
        .expect("build an instance with a provider")
        .0,
    )
}

/// The requests the provider has been handed so far, taken.
fn taken(kept: &Kept) -> Vec<FlushRequest<V>> {
    std::mem::take(&mut *kept.lock().expect("the requests"))
}

/// The only request the provider has been handed.
fn only(kept: &Kept) -> FlushRequest<V> {
    let mut requests = taken(kept);
    assert_eq!(requests.len(), 1, "{requests:?}");
    requests.remove(0)
}

fn succeed(request: FlushRequest<V>) {
    for part in request.domains {
        part.completion.succeed();
    }
}

fn expect_durable(harness: &mut Harness, id: u64) {
    match harness.next_entry() {
        Entry::Durable { through } => assert_eq!(through, harness.epoch(id)),
        other => panic!("expected Durable through {id}, got {other:?}"),
    }
}

fn next_failed(harness: &mut Harness) -> Failed<V> {
    match harness.next_entry() {
        Entry::Failed(failed) => failed,
        other => panic!("expected Failed, got {other:?}"),
    }
}

/// Pop for a while, requiring nothing.
fn nothing_for_a_while(harness: &mut Harness) {
    let started = Instant::now();
    while started.elapsed() < Duration::from_millis(50) {
        assert!(harness.ring.pop().expect("pop").is_none());
        std::thread::sleep(Duration::from_millis(1));
    }
}

/// A provider that answers every domain it is asked about with success, inside `make_durable`.
#[derive(Debug)]
struct Answers(Vec<FlushDomain>);

impl DurabilityProvider<V> for Answers {
    fn domains(&self) -> Vec<FlushDomain> {
        self.0.clone()
    }

    fn make_durable(&mut self, request: FlushRequest<V>) {
        succeed(request);
    }
}

/// A provider that panics when asked anything.
#[derive(Debug)]
struct Panics(Vec<FlushDomain>);

impl DurabilityProvider<V> for Panics {
    fn domains(&self) -> Vec<FlushDomain> {
        self.0.clone()
    }

    fn make_durable(&mut self, _: FlushRequest<V>) {
        panic!("the provider panics");
    }
}

#[test]
fn a_provider_answering_inside_make_durable_makes_the_seal_durable() {
    let temp = temp(&[0; 16]);
    let provider = Answers(vec![domain("log")]);
    let mut harness = instance(&temp, &["log"], Some(Box::new(provider)));
    harness.write(GIVEN, 0, b"log", 1, 1);
    harness.next();
    assert_eq!(harness.seal(1), DurabilityRequest::Submitted);
    expect_durable(&mut harness, 1);
    assert_eq!(harness.state(1), EpochState::Durable);
    harness.finish();
}

#[test]
fn a_request_a_drain_makes_is_made_at_a_pop() {
    // Sealed with its write in flight -- the engine held, so no drain completes the write first --
    // the seal's request is made as the write's completion is drained, off the consumer's thread,
    // and waits for a consumer call (DI-D-44). Here the only calls are pops.
    let temp = temp(&[0; 16]);
    let (provider, kept) = keeper(&["log"]);
    let mut harness = instance(&temp, &["log"], provider);
    let epoch = harness.epoch(1);
    let id = {
        let mut engine = harness.ring.engine();
        let id = engine
            .write(
                GIVEN,
                0,
                SubmissionData::Owned(b"log".to_vec()),
                epoch,
                1,
                WriteOptions::new(),
            )
            .expect("accepted");
        let (answer, _) = engine.make_durable_through(epoch).expect("seal");
        assert_eq!(answer, DurabilityRequest::Submitted);
        id
    };
    harness
        .oracle
        .pushed(id, OpKind::Write { epoch }, 3, 1)
        .expect("the oracle accepts the push");
    harness.oracle.sealed(epoch);
    assert!(
        taken(&kept).is_empty(),
        "the write was in flight at the seal"
    );

    harness.next();
    let started = Instant::now();
    let request = loop {
        if let Some(request) = taken(&kept).pop() {
            break request;
        }
        assert!(started.elapsed() < BOUND, "no pop made the drain's request");
        assert!(harness.ring.pop().expect("pop").is_none());
        std::thread::sleep(Duration::from_millis(1));
    };
    succeed(request);
    expect_durable(&mut harness, 1);
    harness.finish();
}

#[test]
fn a_request_names_each_domain_its_files_and_their_writes() {
    let temps = [temp(&[0; 16]), temp(&[0; 16]), temp(&[0; 16])];
    let files = vec![
        file(2, &temps[0], &["log"]),
        file(3, &temps[1], &["log"]),
        file(4, &temps[2], &["disk"]),
    ];
    let (provider, kept) = keeper(&["log"]);
    let ring = Ring::new(setup(files, Vec::new(), provider))
        .expect("build")
        .0;
    let mut harness = Harness::new(ring);
    let first = harness.write(IoTargetKey(2), 0, b"one", 1, 1);
    let other = harness.write(IoTargetKey(3), 0, b"two", 1, 2);
    let second = harness.write(IoTargetKey(2), 8, b"three", 1, 3);
    harness.write(IoTargetKey(4), 0, b"four", 1, 4);
    harness.next_n(4);
    assert_eq!(harness.seal(1), DurabilityRequest::Submitted);

    let request = only(&kept);
    assert_eq!(request.domains.len(), 1, "disk is the default's");
    let part = &request.domains[0];
    assert_eq!(part.domain.bytes(), b"log");
    let named: Vec<_> = part
        .files
        .iter()
        .map(|f| (f.key, f.writes.clone()))
        .collect();
    assert_eq!(
        named,
        [
            (IoTargetKey(2), vec![first, second]),
            (IoTargetKey(3), vec![other])
        ]
    );
    assert_eq!(harness.state(1), EpochState::Pending);

    succeed(request);
    expect_durable(&mut harness, 1);
    harness.finish();
}

#[test]
fn a_seal_touching_no_served_domain_asks_the_provider_nothing() {
    let temp = temp(&[0; 16]);
    let (provider, kept) = keeper(&["log"]);
    let mut harness = instance(&temp, &["disk"], provider);
    harness.write(GIVEN, 0, b"disk", 1, 1);
    harness.next();
    assert_eq!(harness.seal(1), DurabilityRequest::Submitted);
    expect_durable(&mut harness, 1);
    assert!(taken(&kept).is_empty());
    harness.finish();
}

#[test]
fn an_answer_from_another_thread_sets_the_readiness_signal() {
    let temp = temp(&[0; 16]);
    let (provider, kept) = keeper(&["log"]);
    let mut harness = instance(&temp, &["log"], provider);
    harness.write(GIVEN, 0, b"log", 1, 1);
    harness.next();
    assert_eq!(harness.seal(1), DurabilityRequest::Submitted);
    let request = only(&kept);
    assert!(harness.ring.pop().expect("pop").is_none());

    let signal = harness.ring.readiness().expect("the readiness signal");
    signal.reset().expect("withdraw any stale signal");
    let (woken, wakes) = mpsc::channel();
    let woken = Mutex::new(woken);
    let wait = ThreadpoolWait::new(
        signal.into(),
        move |_| {
            if let Ok(woken) = woken.lock() {
                let _ = woken.send(());
            }
        },
        None,
    )
    .expect("a wait on the signal");
    wait.arm(None);
    std::thread::spawn(move || succeed(request))
        .join()
        .expect("the answering thread");
    let set = wakes.recv_timeout(BOUND);
    wait.stop_and_drain();
    set.expect("the answer made an entry poppable and set the signal");
    expect_durable(&mut harness, 1);
    harness.finish();
}

#[test]
fn a_file_both_answer_for_waits_for_the_provider_after_its_flush() {
    let temp = temp(&[0; 16]);
    let (provider, kept) = keeper(&["log"]);
    let mut harness = instance(&temp, &["log", "disk"], provider);
    harness.write(GIVEN, 0, b"both", 1, 1);
    harness.next();
    assert_eq!(harness.seal(1), DurabilityRequest::Submitted);
    let request = only(&kept);
    nothing_for_a_while(&mut harness);
    assert_eq!(harness.state(1), EpochState::Pending);
    succeed(request);
    expect_durable(&mut harness, 1);
    harness.finish();
}

#[test]
fn a_provider_failure_is_reported_with_its_domain_and_error_and_holds_the_seal() {
    let temp = temp(&[0; 16]);
    let (provider, kept) = keeper(&["log"]);
    let mut harness = instance(&temp, &["log"], provider);
    let write = harness.write(GIVEN, 0, b"log", 1, 1);
    harness.next();
    assert_eq!(harness.seal(1), DurabilityRequest::Submitted);
    let mut request = only(&kept);
    request
        .domains
        .remove(0)
        .completion
        .fail(std::io::Error::from_raw_os_error(IO_DEVICE));

    let failed = next_failed(&mut harness);
    match &failed.cause {
        FailureCause::Provider { domain, error } => {
            assert_eq!(domain.bytes(), b"log");
            assert_eq!(error.raw_os_error(), Some(IO_DEVICE));
        }
        other => panic!("expected a provider failure, got {other:?}"),
    }
    let suspects: Vec<_> = failed.suspect.writes().iter().map(|w| w.op).collect();
    assert_eq!(suspects, [write]);
    assert_eq!(harness.state(1), EpochState::Blocked(failed.id));
    assert!(harness.ring.pop().expect("pop").is_none(), "one answer");
    harness.finish();
}

#[test]
fn a_completion_dropped_unanswered_is_reported_as_abandoned() {
    let temp = temp(&[0; 16]);
    let (provider, kept) = keeper(&["log"]);
    let mut harness = instance(&temp, &["log"], provider);
    harness.write(GIVEN, 0, b"log", 1, 1);
    harness.next();
    assert_eq!(harness.seal(1), DurabilityRequest::Submitted);
    drop(only(&kept));

    let failed = next_failed(&mut harness);
    assert!(
        matches!(&failed.cause, FailureCause::ProviderAbandoned { domain } if domain.bytes() == b"log"),
        "{:?}",
        failed.cause
    );
    assert_eq!(harness.state(1), EpochState::Blocked(failed.id));
    harness.finish();
}

#[test]
fn a_panic_in_the_provider_reaches_the_consumers_call_and_abandons_what_it_dropped() {
    let temp = temp(&[0; 16]);
    let provider = Panics(vec![domain("log")]);
    let mut harness = instance(&temp, &["log"], Some(Box::new(provider)));
    harness.write(GIVEN, 0, b"log", 1, 1);
    harness.next();
    let at = harness.default.at(1);
    let sealing = panic::catch_unwind(AssertUnwindSafe(|| harness.ring.make_durable_through(at)));
    assert!(sealing.is_err(), "the provider's panic propagates");
    harness.oracle.sealed(harness.epoch(1));

    let failed = next_failed(&mut harness);
    assert!(
        matches!(&failed.cause, FailureCause::ProviderAbandoned { domain } if domain.bytes() == b"log"),
        "{:?}",
        failed.cause
    );
    assert_eq!(
        harness.seal(1),
        DurabilityRequest::AlreadySealed(EpochState::Blocked(failed.id)),
        "the instance is still usable"
    );
    harness.finish();
}

// The routing provider as the instance's provider (DI-12.2.4, DI-D-81, DI-D-83): two providers
// composed by a router, the log's and the disk's, each a keeper the test answers for.

const LOG: IoTargetKey = IoTargetKey(2);
const DISK: IoTargetKey = IoTargetKey(3);

/// A keeper serving `domains`, by itself, so a router can register it.
fn kept_by(domains: &[&str]) -> (Keeper, Kept) {
    let kept = Arc::new(Mutex::new(Vec::new()));
    let keeper = Keeper {
        domains: domains.iter().map(|name| domain(name)).collect(),
        kept: Arc::clone(&kept),
    };
    (keeper, kept)
}

/// An instance over `files`, whose provider is a router over a keeper serving `log` and one serving
/// `disk`, in that order; and what each keeps.
fn routed(files: Vec<TargetSetup>) -> (Harness, Kept, Kept) {
    let (log, log_kept) = kept_by(&["log"]);
    let (disk, disk_kept) = kept_by(&["disk"]);
    let router = RoutingProvider::builder()
        .register(log)
        .register(disk)
        .build()
        .expect("no domain twice");
    let ring = Ring::new(setup(files, Vec::new(), Some(Box::new(router))))
        .expect("build an instance with a router")
        .0;
    (Harness::new(ring), log_kept, disk_kept)
}

/// The files `requested` names, each with its writes.
fn named(request: &FlushRequest<V>) -> Vec<(String, IoTargetKey, Vec<OpId>)> {
    request
        .domains
        .iter()
        .flat_map(|part| {
            let name = String::from_utf8_lossy(part.domain.bytes()).into_owned();
            part.files
                .iter()
                .map(move |file| (name.clone(), file.key, file.writes.clone()))
        })
        .collect()
}

#[test]
fn a_seal_touching_both_routed_providers_asks_each_once_for_its_own_domain() {
    let temps = [temp(&[0; 16]), temp(&[0; 16])];
    let (mut harness, log_kept, disk_kept) = routed(vec![
        file(LOG.0, &temps[0], &["log"]),
        file(DISK.0, &temps[1], &["disk"]),
    ]);
    let log_write = harness.write(LOG, 0, b"log", 1, 1);
    let disk_write = harness.write(DISK, 0, b"disk", 1, 2);
    harness.next_n(2);
    assert_eq!(harness.seal(1), DurabilityRequest::Submitted);

    let log = only(&log_kept);
    let disk = only(&disk_kept);
    assert_eq!(named(&log), [("log".to_owned(), LOG, vec![log_write])]);
    assert_eq!(named(&disk), [("disk".to_owned(), DISK, vec![disk_write])]);

    succeed(log);
    nothing_for_a_while(&mut harness);
    assert_eq!(
        harness.state(1),
        EpochState::Pending,
        "the disk has not answered"
    );
    succeed(disk);
    expect_durable(&mut harness, 1);
    harness.finish();
}

#[test]
fn a_file_whose_domains_span_routed_providers_is_durable_only_when_every_one_succeeds() {
    // Its write is in each provider's call: the router merges nothing across providers, so a
    // file whose coupled domains are split between them costs each a flush (DI-2.12 point 1).
    let temp = temp(&[0; 16]);
    let (mut harness, log_kept, disk_kept) = routed(vec![file(GIVEN.0, &temp, &["log", "disk"])]);
    let write = harness.write(GIVEN, 0, b"both", 1, 1);
    harness.next();
    assert_eq!(harness.seal(1), DurabilityRequest::Submitted);

    let log = only(&log_kept);
    let disk = only(&disk_kept);
    assert_eq!(named(&log), [("log".to_owned(), GIVEN, vec![write])]);
    assert_eq!(named(&disk), [("disk".to_owned(), GIVEN, vec![write])]);

    succeed(disk);
    nothing_for_a_while(&mut harness);
    assert_eq!(
        harness.state(1),
        EpochState::Pending,
        "the log has not answered"
    );
    succeed(log);
    expect_durable(&mut harness, 1);
    harness.finish();
}

#[test]
fn one_routed_provider_failing_fails_only_the_files_in_its_domains() {
    let temps = [temp(&[0; 16]), temp(&[0; 16])];
    let (mut harness, log_kept, disk_kept) = routed(vec![
        file(LOG.0, &temps[0], &["log"]),
        file(DISK.0, &temps[1], &["disk"]),
    ]);
    let log_write = harness.write(LOG, 0, b"log", 1, 1);
    harness.write(DISK, 0, b"disk", 1, 2);
    harness.next_n(2);
    assert_eq!(harness.seal(1), DurabilityRequest::Submitted);

    succeed(only(&disk_kept));
    let mut log = only(&log_kept);
    log.domains
        .remove(0)
        .completion
        .fail(std::io::Error::from_raw_os_error(IO_DEVICE));

    let failed = next_failed(&mut harness);
    match &failed.cause {
        FailureCause::Provider { domain, error } => {
            assert_eq!(domain.bytes(), b"log");
            assert_eq!(error.raw_os_error(), Some(IO_DEVICE));
        }
        other => panic!("expected the log provider's failure, got {other:?}"),
    }
    let suspects: Vec<_> = failed.suspect.writes().iter().map(|w| w.op).collect();
    assert_eq!(
        suspects,
        [log_write],
        "the disk's file is not in the log's domain"
    );
    assert_eq!(harness.state(1), EpochState::Blocked(failed.id));
    harness.finish();
}

#[test]
fn a_file_whose_domains_span_routed_providers_fails_when_either_fails() {
    let temp = temp(&[0; 16]);
    let (mut harness, log_kept, disk_kept) = routed(vec![file(GIVEN.0, &temp, &["log", "disk"])]);
    let write = harness.write(GIVEN, 0, b"both", 1, 1);
    harness.next();
    assert_eq!(harness.seal(1), DurabilityRequest::Submitted);

    succeed(only(&log_kept));
    let mut disk = only(&disk_kept);
    disk.domains
        .remove(0)
        .completion
        .fail(std::io::Error::from_raw_os_error(IO_DEVICE));

    let failed = next_failed(&mut harness);
    assert!(
        matches!(&failed.cause, FailureCause::Provider { domain, .. } if domain.bytes() == b"disk"),
        "{:?}",
        failed.cause
    );
    let suspects: Vec<_> = failed.suspect.writes().iter().map(|w| w.op).collect();
    assert_eq!(suspects, [write]);
    assert_eq!(harness.state(1), EpochState::Blocked(failed.id));
    harness.finish();
}

#[test]
fn a_domain_no_routed_provider_serves_is_the_defaults() {
    let temp = temp(&[0; 16]);
    let (mut harness, log_kept, disk_kept) = routed(vec![file(GIVEN.0, &temp, &["other"])]);
    harness.write(GIVEN, 0, b"other", 1, 1);
    harness.next();
    assert_eq!(harness.seal(1), DurabilityRequest::Submitted);
    expect_durable(&mut harness, 1);
    assert!(taken(&log_kept).is_empty());
    assert!(taken(&disk_kept).is_empty());
    harness.finish();
}

#[test]
fn a_panic_in_a_routed_provider_abandons_its_domains_and_those_not_yet_handed_on() {
    let temps = [temp(&[0; 16]), temp(&[0; 16])];
    let (disk, disk_kept) = kept_by(&["disk"]);
    let router = RoutingProvider::builder()
        .register(Panics(vec![domain("log")]))
        .register(disk)
        .build()
        .expect("no domain twice");
    let ring = Ring::new(setup(
        vec![
            file(LOG.0, &temps[0], &["log"]),
            file(DISK.0, &temps[1], &["disk"]),
        ],
        Vec::new(),
        Some(Box::new(router)),
    ))
    .expect("build an instance with a router")
    .0;
    let mut harness = Harness::new(ring);
    harness.write(LOG, 0, b"log", 1, 1);
    harness.write(DISK, 0, b"disk", 1, 2);
    harness.next_n(2);
    let at = harness.default.at(1);
    let sealing = panic::catch_unwind(AssertUnwindSafe(|| harness.ring.make_durable_through(at)));
    assert!(
        sealing.is_err(),
        "the provider's panic propagates through the router"
    );
    harness.oracle.sealed(harness.epoch(1));
    assert!(taken(&disk_kept).is_empty(), "the disk was never called");

    let mut abandoned = Vec::new();
    for _ in 0..2 {
        match next_failed(&mut harness).cause {
            FailureCause::ProviderAbandoned { domain } => abandoned.push(domain),
            other => panic!("expected an abandoned domain, got {other:?}"),
        }
    }
    abandoned.sort_by(|a, b| a.bytes().cmp(b.bytes()));
    assert_eq!(abandoned, [domain("disk"), domain("log")]);
    harness.finish();
}