win-durable-ioring

A ring for Windows file I/O with durability scheduled into its work: the Windows driver for durable-ioring, built on windows-ioring-sys's IoRing. A consumer submits reads and writes as it would to the ring, tags each write with an epoch, asks for durability with make_durable_through(n), and learns -- in the same completion stream as its operations -- when everything through n has become durable, or that it has failed and which writes are suspect as a result.

What is here, and what is not

Here (DI-D-50): Dioring, the threaded driver that implements the contract over the Windows backend; the backend and its file target (windows_backend), the thread-pool and pumped and signalled delivery of its completions, the readiness Event and the harness check of it (readiness), the interrupt-time clock the driver stamps with, and a way to read an HRESULT out of an io::Error (error_code_of). It forbids unsafe code: a Win32 call no lower crate wraps safely is added to that crate.

Not here: the contract, the core that decides when writes are durable, the target and backend traits, the in-memory backend and the conformance oracle. Those are durable-ioring, which has no Windows dependency, so another platform's driver can be built on it as this one is. The specification is that crate's CONTRACT.md; its spelling as Rust types is API.md. This crate re-exports the portable crate's API, so a consumer needs only this one.

A blocking push is not an API of either crate: a refused push names its cause, and the loop that waits it out is a maintained example, examples/blocking_push/, to use as it stands or adapt (DI-D-56).

Diagnostics

An instance reports what it waits on as pairs of events, a begin and an end sharing an id -- a seal, from its being asked for to its answer; each flush of a file or a domain; and a wait for one of its locks -- and its inconsistencies, to a DiagnosticSink given in Setup::diagnostics. How long each took is the gap between its begin and its end, which the reader measures; the instance times nothing. Given none, it writes them to ETW, to the provider durable-ioring-events (durable-ioring-events), through EtwDiagnostics. That costs a test of whether a session is listening when none is, and no event is built; the first such instance registers the provider for the process, and the last gives it back. Any other sink is given in its place: NoDiagnostics for none, StderrDiagnostics or DebugOutputDiagnostics for text, or FanOutDiagnostics to send events to several. A reader of the trace needs the provider's manifest to see names and fields; the instance does not.

DeliveryMode::Pool, DeliveryMode::Signalled and EntryDelivery::new take windows-threadpool-sys's CallbackEnviron, so that crate's types are part of this crate's public API. A minor release of windows-threadpool-sys that changes those types is a minor release of this crate too, in the 0.x series: 0.3 of this crate pairs with 0.3 of windows-threadpool-sys.

Several durability providers

An instance holds the built-in provider, which flushes each file through its own ring, and at most one more, given in Setup::provider: a DurabilityProvider serving the flush domains it declares, which answers for those domains of each seal. Several are composed by RoutingProvider, itself a provider. Each provider registered with it serves exactly the domains its own domains() returns, read once when the router is built; a domain claimed twice refuses the build, naming both registrations. A seal's work is split by domain, and each provider with a part is called once, in registration order; its answers go straight back to the instance. A file declared with domains of two providers is in both calls and is durable when both have succeeded; a domain no provider serves stays the built-in provider's. The rules are in API.md section 15 and DI-D-81 to DI-D-83.

use win_durable_ioring::*;

/// A provider for one device, serving the domains it was built with. A real one flushes the
/// files each part names, by its own means, before it answers.
#[derive(Debug)]
struct Device(Vec<FlushDomain>);

impl DurabilityProvider<WindowsIds<u64>> for Device {
    fn domains(&self) -> Vec<FlushDomain> {
        self.0.clone()
    }

    fn make_durable(&mut self, request: FlushRequest<WindowsIds<u64>>) {
        for part in request.domains {
            part.completion.succeed();
        }
    }
}

let log = FlushDomain::new(b"log-disk".to_vec());
let data = FlushDomain::new(b"data-disk".to_vec());
let router = RoutingProvider::builder()
    .register(Device(vec![log.clone()]))
    .register(Device(vec![data.clone()]))
    .build()?;
assert_eq!(router.domains(), [log.clone(), data]);

let (ring, _default) = Dioring::<Vec<u8>>::new(Setup {
    submission_queue_size: 16,
    completion_queue_size: 16,
    files: Vec::new(),
    buffers: Vec::new(),
    provider: Some(Box::new(router)),
    diagnostics: None,
    delivery: DeliveryMode::Pool(None),
})
.expect("an instance");
drop(ring);

// The same domain from two providers refuses the router, naming both registrations.
let refused = RoutingProvider::<WindowsIds<u64>>::builder()
    .register(Device(vec![log.clone()]))
    .register(Device(vec![log.clone()]))
    .build()
    .expect_err("the log twice");
assert_eq!((refused.domain, refused.first, refused.second), (log, 0, 1));
# Ok::<(), Box<dyn std::error::Error>>(())

Opening a file

A file target is made from a file this crate opened, and from nothing else (DI-D-60). How a handle was opened decides what its target can claim, and the only call that reports it is one this repository does not make, so the open mode is recorded by the type that opens the file instead of asked of a handle afterwards. OpenedFile<Cached> is a file opened for cached I/O: it claims BestEffort and needs no alignment. OpenedFile<Unbuffered> is opened with no intermediate buffering: it claims SharedDevice, needs a sector's alignment (4096 unless told less with with_sector_size), and offers per-write write-through. A consumer that needs the stronger warranty asks for OpenedFile<Unbuffered> in its own signature, and a cached file cannot be passed. A handle opened elsewhere cannot be given to a target in this version.

The data access of a handle is its type. FileOptions::new() asks for none, read() and write() add the right to read and to write, and the file an open makes is an OpenedFile<B, A> whose A is NoData, ReadOnly, WriteOnly or ReadWrite. A target needs a data right, so a file opened with NoData cannot be made into one, and the program does not compile. OpenedDisk and OpenedVolume carry the same parameter, so either can be opened for reading only. The other rights an open can ask for, the attribute, delete and security rights, are a FileRights set given to FileOptions::rights; they gate calls a consumer makes on the handle it is lent and not anything this crate does. There is no constructor from a raw mask and no right for appending only, which would land every write at the end of the file whatever offset the core names, so it cannot be asked for. What the file's security allows is the kernel's to say.

FileOptions::share takes a ShareMode, a set of READ, WRITE and DELETE naming what other opens of the file may request while this one is open; a file is opened sharing all three unless it is told otherwise, and ShareMode::NONE is an exclusive open. An open that the sharing of either side does not allow fails as a SharingViolation, an io::ErrorKind::ResourceBusy error whose cause carries what this open asked for. CreateFileW reports both sides' conflicts as one error and no documented call reports what the other open holds, so it cannot say which side conflicted. The kernel's own error is its source.

FileOptions::random_access and FileOptions::sequential_scan are the two cache hints (FILE_FLAG_RANDOM_ACCESS and FILE_FLAG_SEQUENTIAL_SCAN). They are passed to an unbuffered file as well as a cached one, never refused, and change nothing a target claims: the class, the alignment and what it offers follow from the buffering mode alone. What the CreateFileW documentation says about them on an unbuffered file can be read two ways, and the methods' own documentation says what was measured.

FileOptions::overlapped sets FILE_FLAG_OVERLAPPED. The ring takes such a handle and gives the result for each operation that it gives without the flag; what the flag changes is when a completion is there and, for an unbuffered file, the order a batch's completions come in. The handle the open lends is the same overlapped handle, so I/O a caller makes on it directly, without an offset, is I/O on a handle for which CreateFileW says the file pointer is not maintained.

The path is a Wtf16String, handed to Win32 as the caller gave it: nothing lengthens, shortens or refuses it, so a path past MAX_PATH without a \\?\ prefix, and an empty one, are what CreateFileW makes of them. Convert a Path with Wtf16String::from_os_str, which loses nothing.

FileOptions::disposition takes a Disposition: CreateNew, OpenExisting (where the options start) or OpenAlways, named as CreateFileW names them. CREATE_ALWAYS and TRUNCATE_EXISTING, the two that empty a file that exists, are not offered in this version; a caller empties a file through the handle the open lends, with File::set_len(0). FileOptions::open_reparse_point sets FILE_FLAG_OPEN_REPARSE_POINT, which no disposition implies: without it a link is followed, including by CreateNew.

OpenedFile::open blocks for as long as CreateFileW does. OpenedFile::request builds an OpenRequest and opens nothing, fixing a relative path to the current directory as it is then, and OpenRequest::perform makes the call on whatever thread the caller chooses. A request is Send, and performing it again opens the file again. The refusal of an unbuffered file on a remote device is made in perform, on the opened handle from its final path, so a file the open created is still there when it is refused.

A file the open creates is given FileOptions::attributes, a FileAttributes set of ARCHIVE, HIDDEN, OFFLINE, READONLY, SYSTEM and TEMPORARY (FILE_ATTRIBUTE_NORMAL, the absence of attributes, is the empty set, and FILE_ATTRIBUTE_ENCRYPTED is not offered), and OpenRequest::with_security_descriptor, a SecurityDescriptor built from SDDL, which the request owns and frees. The descriptor is on the request and not in the options because it owns memory and cannot be copied, as the options can. Neither applies to a file that already exists. Handles are never inheritable, and there is no option to make one so.

use win_durable_ioring::windows_backend::{
    Cached, Disposition, FileAttributes, FileOptions, OpenedFile, SecurityDescriptor,
};
use win_durable_ioring::Wtf16String;

let path = std::env::temp_dir().join(format!("durable-ioring-create-{}", std::process::id()));
let options = FileOptions::new()
    .read()
    .write()
    .disposition(Disposition::CreateNew)
    .attributes(FileAttributes::HIDDEN);
let descriptor = SecurityDescriptor::from_sddl(&Wtf16String::from("D:P(A;;GA;;;WD)"))?;

let created = OpenedFile::<Cached>::request(Wtf16String::from_os_str(path.as_os_str()), options)?
    .with_security_descriptor(descriptor)
    .perform()?;
drop(created);
# std::fs::remove_file(&path)?;
# Ok::<(), Box<dyn std::error::Error>>(())

use win_durable_ioring::windows_backend::{
    Cached, Disposition, FileOptions, OpenedFile, Unbuffered, Win32FileTarget,
};
use win_durable_ioring::{IoTarget, WarrantyClass, Wtf16String};

let path = std::env::temp_dir().join(format!("durable-ioring-open-{}", std::process::id()));
std::fs::write(&path, [0u8; 8192])?;
let options = FileOptions::new().read().write();
// The open takes the path as the caller gave it, a `Wtf16String`: converting loses nothing.
let name = || Wtf16String::from_os_str(path.as_os_str());

let cached = Win32FileTarget::new(OpenedFile::<Cached>::open(name(), options)?);
assert_eq!(cached.warranty(), WarrantyClass::BestEffort);

let unbuffered = Win32FileTarget::new(OpenedFile::<Unbuffered>::open(name(), options)?);
assert_eq!(unbuffered.warranty(), WarrantyClass::SharedDevice);
assert!(unbuffered.capabilities().write_through());

// Built here, and made on the thread that may block.
let request = OpenedFile::<Cached>::request(name(), options.disposition(Disposition::OpenAlways))?;
let opened = std::thread::spawn(move || request.perform()).join().expect("the thread")?;
drop(opened);
# std::fs::remove_file(&path)?;
# Ok::<(), std::io::Error>(())

An unbuffered file is refused on a remote device, where the claim cannot hold. The handle is asked where it is after the open, so a link is judged by the handle it opens: followed, the file it leads to; opened with open_reparse_point, the link itself.

By identity

A file has an identity, FileId: its volume's serial number and a 128-bit identifier, read from any handle to it. OpenedFile::open_by_id opens it again by that, or by a 64-bit file ID or an object ID, on the volume of a hint, which is any handle on that volume (DI-D-70). Its options are ByIdOptions, since OpenFileById takes no disposition and no attributes. The call never sees a FileId's volume, so the file it opens is asked its identity, and one from another volume is refused as an IdentityMismatch. An identifier can name a directory, which opens, and a target made from it faults every read and write (DI-D-72).

use win_durable_ioring::windows_backend::{ByIdOptions, Cached, FileOptions, OpenedFile};
use win_durable_ioring::Wtf16String;

let path = std::env::temp_dir().join(format!("durable-ioring-by-id-{}", std::process::id()));
std::fs::write(&path, b"by identity")?;
let file = OpenedFile::<Cached, _>::open(
    Wtf16String::from_os_str(path.as_os_str()),
    FileOptions::new().read(),
)?;
let id = file.file_id()?;

// Any handle on the volume is a hint: here, another file's, opened by std.
let neighbour = path.with_extension("neighbour");
std::fs::write(&neighbour, b"")?;
let hint = std::fs::File::open(&neighbour)?;
let again = OpenedFile::<Cached, _>::open_by_id(hint, id, ByIdOptions::new().read())?;
assert_eq!(again.file_id()?, id);
# drop((file, again));
# std::fs::remove_file(&path)?;
# std::fs::remove_file(&neighbour)?;
# Ok::<(), std::io::Error>(())

Reopening

OpenedFile::reopen makes a new handle to what a handle already names, through ReOpenFile, in the buffering mode and with the access its type asks for (DI-D-71): cached to unbuffered and back, or a read-only handle for a scrubber beside the writer. The original is any handle, this crate's or one opened elsewhere, and the result is judged by the same checks as an open by path. Its options are ReopenOptions. The reopen's sharing is judged against every handle still open to the file, the original among them, so a reopen of a file its original opened without sharing is refused as a SharingViolation.

use win_durable_ioring::windows_backend::{
    Cached, FileOptions, OpenedFile, ReadOnly, ReopenOptions, Unbuffered,
};
use win_durable_ioring::Wtf16String;

let path = std::env::temp_dir().join(format!("durable-ioring-reopen-{}", std::process::id()));
std::fs::write(&path, [0u8; 4096])?;
let writer = OpenedFile::<Cached, _>::open(
    Wtf16String::from_os_str(path.as_os_str()),
    FileOptions::new().read().write(),
)?;

let scrubber = OpenedFile::<Unbuffered, ReadOnly>::reopen(
    writer.handle().clone(),
    ReopenOptions::new().read(),
)?;
assert_eq!(scrubber.file_id()?, writer.file_id()?);
# drop((writer, scrubber));
# std::fs::remove_file(&path)?;
# Ok::<(), std::io::Error>(())

A raw volume

A raw volume is a target too (DI-D-61): one handle, one commit per volume, and regions are offsets the consumer allocates inside it, not targets of their own. OpenedVolume opens it by a path that can only name a volume, a drive letter or a volume GUID, and finds what the backing must know for itself with documented device controls (win-volume-sys): the disks it sits on, which become its flush domains (disk_flush_domain names a disk, so a consumer can declare the same domain for a file on it), and the sector it is addressed in, which is the alignment of its operations. A volume that is remote, or not a volume, is refused. It claims SharedDevice when opened unbuffered and BestEffort when opened cached, and never Complete: a lock keeps other handles off the volume, not other writers off its disks. Opening one for writing needs an administrator, and VolumeOptions::lock takes the volume lock, which a write to a volume with a file system on it needs and which fails if anything else has the volume open.

use win_durable_ioring::windows_backend::{
    OpenedVolume, ReadWrite, Unbuffered, VolumeOptions, VolumePath, Win32FileTarget,
};

let path = VolumePath::drive('E')?;
// The data access is named in the type: a volume opened for reading only is `ReadOnly`.
let volume =
    OpenedVolume::<Unbuffered, ReadWrite>::open(&path, VolumeOptions::new().lock(true))?;
let target = Win32FileTarget::from_volume(volume);
# Ok::<(), Box<dyn std::error::Error>>(())

A whole disk

OpenedDisk opens a physical disk by its number, \\.\PhysicalDriveN, rather than a volume on it (DI-D-62). The disk is its own flush domain, named by disk_flush_domain as a volume on it would name it, and its operations are aligned to its sector, which a device control reports. It claims SharedDevice when opened unbuffered and BestEffort when opened cached, never Complete, for the reason a volume does not. Windows documents that it blocks a direct write to a sector a mounted volume covers, so a disk with volumes on it is written through them, or has them locked or is taken offline first; the tests here use a disk with no partition. Opening one for writing needs an administrator. A disk and a volume are opened sharing reading, writing and deletion unless DiskOptions::share or VolumeOptions::share takes a ShareMode otherwise, and a conflict is the same SharingViolation a file reports.

use win_durable_ioring::windows_backend::{
    DiskOptions, DiskPath, OpenedDisk, ReadOnly, Unbuffered, Win32FileTarget,
};

// The data access is named in the type: this one is opened for reading only.
let disk = OpenedDisk::<Unbuffered, ReadOnly>::open(DiskPath::number(3), DiskOptions::new())?;
let target = Win32FileTarget::from_disk(disk);
# Ok::<(), std::io::Error>(())

Worked examples

Each example is dioring running the trace in its comments, compiled and run as a test, so a change that breaks one breaks the build. In the comments, -> is a call the consumer makes and <- an entry it pops. Every write is in the default lineage, so the lineage is not shown; construction returns a handle to it, and tag(n) is its epoch n as a write or a seal takes it, epoch(n) as a query or an entry names it. A commit failure cannot be provoked on a healthy machine through a Win32 file, so each example arms one with dioring's fault seam, fail_next_flush, which makes the next flush of a file report the error given whatever the device answered. Setting up the instance, and a pop_until that pops until an entry it is looking for, are hidden.

Healing a failure

# use std::time::{Duration, Instant};
# use win_durable_ioring::*;
# type Ring = Dioring<Vec<u8>>;
# const FILE: IoTargetKey = IoTargetKey(1);
# fn pop_until(ring: &mut Ring, stop: impl Fn(&EntryOf<Ring>) -> bool) -> EntryOf<Ring> {
#     let started = Instant::now();
#     loop {
#         match ring.pop().expect("pop") {
#             Some(entry) if stop(&entry) => return entry,
#             Some(_) => {}
#             None => {
#                 assert!(started.elapsed() < Duration::from_secs(10), "nothing arrived");
#                 std::thread::sleep(Duration::from_millis(1));
#             }
#         }
#     }
# }
# let path = std::env::temp_dir().join(format!("durable-ioring-heal-{}", std::process::id()));
# std::fs::write(&path, [0u8; 64])?;
# let file = windows_backend::OpenedFile::<windows_backend::Cached>::open(
#     win_durable_ioring::Wtf16String::from_os_str(path.as_os_str()),
#     windows_backend::FileOptions::new().read().write(),
# )?;
# let (mut ring, lineage) = Ring::new(Setup {
#     submission_queue_size: 64,
#     completion_queue_size: 64,
#     files: vec![TargetSetup::new(
#         FILE,
#         windows_backend::Win32FileTarget::new(file),
#     )],
#     buffers: Vec::new(),
#     provider: None,
#     diagnostics: None,
#     delivery: DeliveryMode::Pool(None),
# })
# .expect("an instance");
let tag = |n: u64| lineage.at(n); // what a write or a seal takes
let epoch = |n: u64| Epoch::new(lineage.lineage(), n); // what a query or an entry names

// -> write(tag 41, A) ; write(tag 41, B) ; write(tag 41, C)
for offset in [0, 8, 16] {
    ring.write(FILE, offset, vec![1; 8], tag(41), ()).expect("a write");
}
ring.fail_next_flush(FILE, 1117); // ERROR_IO_DEVICE, for the flush the seal issues
// -> make_durable_through(41)                 // seals everything <= 41
ring.make_durable_through(tag(41)).expect("a seal");
// <- WriteDone(A) ; WriteDone(B) ; WriteDone(C)
// <- Failed { failure: F1, cause: Commit, suspect: [A@41, B@41, C@41] }
let Entry::Failed(f1) = pop_until(&mut ring, |e| matches!(e, Entry::Failed(_))) else {
    unreachable!()
};
assert!(matches!(f1.cause, FailureCause::Commit { .. }));
assert_eq!(f1.suspect.writes().len(), 3);
assert_eq!(ring.durable_through(lineage.lineage()), Ok(None)); // nothing is durable yet
// -> write(tag 41, A')                        // refused: 41 is sealed (guarantee 6)
let refused = ring.write(FILE, 0, vec![2; 8], tag(41), ()).expect_err("41 is sealed");
assert!(matches!(refused.reason, PushRefusal::Sealed { .. }));
// -> write(tag 42, A') ; write(tag 42, B') ; write(tag 42, C')
for offset in [0, 8, 16] {
    ring.write(FILE, offset, vec![2; 8], tag(42), ()).expect("a write");
}
// -> heal(F1)                                 // "I have re-issued what I need"
ring.resolve(vec![(f1.token, Resolution::Heal)]).expect("heal");
// -> make_durable_through(42)                 // the first after the heal reaching the re-issue
ring.make_durable_through(tag(42)).expect("a seal");
// <- WriteDone(A') ; WriteDone(B') ; WriteDone(C')
// <- Durable { through: 42 }                  // F1 healed; 41 and 42 durable
pop_until(&mut ring, |e| matches!(e, Entry::Durable { through } if *through == epoch(42)));
assert_eq!(ring.epoch_state(epoch(41)), Ok(EpochState::Durable));
assert!(ring.failures().is_empty());
# drop(ring);
# std::fs::remove_file(&path)?;
# Ok::<(), std::io::Error>(())

Without the heal, the last entry is Blocked { through: 42, by: F1 }: 42's own commit succeeded, but the high-water mark cannot pass the unresolved failure at 41.

Failures and epochs are many-to-many

P is pushed before the seal, rather than after it as a consumer might, so that it is certainly pushed before the failure is observed: the seal's commit is issued as soon as A completes.

# use std::time::{Duration, Instant};
# use win_durable_ioring::*;
# type Ring = Dioring<Vec<u8>>;
# const FILE: IoTargetKey = IoTargetKey(1);
# fn pop_until(ring: &mut Ring, stop: impl Fn(&EntryOf<Ring>) -> bool) -> EntryOf<Ring> {
#     let started = Instant::now();
#     loop {
#         match ring.pop().expect("pop") {
#             Some(entry) if stop(&entry) => return entry,
#             Some(_) => {}
#             None => {
#                 assert!(started.elapsed() < Duration::from_secs(10), "nothing arrived");
#                 std::thread::sleep(Duration::from_millis(1));
#             }
#         }
#     }
# }
# let path = std::env::temp_dir().join(format!("durable-ioring-many-{}", std::process::id()));
# std::fs::write(&path, [0u8; 64])?;
# let file = windows_backend::OpenedFile::<windows_backend::Cached>::open(
#     win_durable_ioring::Wtf16String::from_os_str(path.as_os_str()),
#     windows_backend::FileOptions::new().read().write(),
# )?;
# let (mut ring, lineage) = Ring::new(Setup {
#     submission_queue_size: 64,
#     completion_queue_size: 64,
#     files: vec![TargetSetup::new(
#         FILE,
#         windows_backend::Win32FileTarget::new(file),
#     )],
#     buffers: Vec::new(),
#     provider: None,
#     diagnostics: None,
#     delivery: DeliveryMode::Pool(None),
# })
# .expect("an instance");
let tag = |n: u64| lineage.at(n); // what a write or a seal takes
let epoch = |n: u64| Epoch::new(lineage.lineage(), n); // what a query or an entry names
let ops = |set: &SuspectSet<WindowsIds<u64>>| set.writes().iter().map(|w| w.op).collect::<Vec<_>>();

// -> write(tag 41, A) ; write(tag 43, P)      // 43 is open
let a = ring.write(FILE, 0, vec![1; 8], tag(41), ()).expect("a write");
let p = ring.write(FILE, 8, vec![1; 8], tag(43), ()).expect("a write");
ring.fail_next_flush(FILE, 1117);
// -> make_durable_through(41)                 // seals 41
ring.make_durable_through(tag(41)).expect("a seal");
// <- WriteDone(A) ; WriteDone(P)
// <- Failed { failure: F1, cause: Commit, suspect: [A@41, P@43] }
let Entry::Failed(f1) = pop_until(&mut ring, |e| matches!(e, Entry::Failed(_))) else {
    unreachable!()
};
assert_eq!(ops(&f1.suspect), [a, p]); // one failure, two epochs: sealed 41 and open 43
// -> write(tag 43, R)                         // still allowed: 43 is open
let r = ring.write(FILE, 16, vec![1; 8], tag(43), ()).expect("a write");
// <- WriteDone(R)
pop_until(&mut ring, |e| matches!(e, Entry::Op(done) if done.id == r));
// -> import_failure()                         // the consumer learned of an outside failure
let f2_id = ring.import_failure(ImportScope::All).expect("the whole instance").failure;
// <- Failed { failure: F2, cause: Imported, suspect: [A@41, P@43, R@43] }
let Entry::Failed(f2) = pop_until(&mut ring, |e| matches!(e, Entry::Failed(_))) else {
    unreachable!()
};
assert_eq!(f2.id, f2_id);
assert_eq!(ops(&f2.suspect), [a, p, r]); // epoch 43 now sits in F1 and F2

// The first ending: heal both. Re-issue A, P and R under tag 44, resolve both in one atomic
// call, and seal 44.
for offset in [0, 8, 16] {
    ring.write(FILE, offset, vec![2; 8], tag(44), ()).expect("a write");
}
ring.resolve(vec![(f1.token, Resolution::Heal), (f2.token, Resolution::Heal)])
    .expect("heal both");
ring.make_durable_through(tag(44)).expect("a seal");
// <- Durable { through: 44 }                  // 41 and 43 passed once both failures were resolved
pop_until(&mut ring, |e| matches!(e, Entry::Durable { through } if *through == epoch(44)));
assert_eq!(ring.epoch_state(epoch(41)), Ok(EpochState::Durable));
assert_eq!(ring.epoch_state(epoch(43)), Ok(EpochState::Durable));
# drop(ring);
# std::fs::remove_file(&path)?;
# Ok::<(), std::io::Error>(())

The second ending: heal F1, abandon F2. Epochs 41 and 43 both sit in F2, so both become abandoned -- one abandonment is enough -- even though F1 was healed. The high-water mark moves past them once every failure containing them is resolved: F2 at once, and F1 when its heal takes effect, at the first seal made after it that reaches the re-issue. Any operation gated on 41 or 43 fails as never issued.

# use std::time::{Duration, Instant};
# use win_durable_ioring::*;
# type Ring = Dioring<Vec<u8>>;
# const FILE: IoTargetKey = IoTargetKey(1);
# fn pop_until(ring: &mut Ring, stop: impl Fn(&EntryOf<Ring>) -> bool) -> EntryOf<Ring> {
#     let started = Instant::now();
#     loop {
#         match ring.pop().expect("pop") {
#             Some(entry) if stop(&entry) => return entry,
#             Some(_) => {}
#             None => {
#                 assert!(started.elapsed() < Duration::from_secs(10), "nothing arrived");
#                 std::thread::sleep(Duration::from_millis(1));
#             }
#         }
#     }
# }
# let path = std::env::temp_dir().join(format!("durable-ioring-abandon-{}", std::process::id()));
# std::fs::write(&path, [0u8; 64])?;
# let file = windows_backend::OpenedFile::<windows_backend::Cached>::open(
#     win_durable_ioring::Wtf16String::from_os_str(path.as_os_str()),
#     windows_backend::FileOptions::new().read().write(),
# )?;
# let (mut ring, lineage) = Ring::new(Setup {
#     submission_queue_size: 64,
#     completion_queue_size: 64,
#     files: vec![TargetSetup::new(
#         FILE,
#         windows_backend::Win32FileTarget::new(file),
#     )],
#     buffers: Vec::new(),
#     provider: None,
#     diagnostics: None,
#     delivery: DeliveryMode::Pool(None),
# })
# .expect("an instance");
# let tag = |n: u64| lineage.at(n);
# let epoch = |n: u64| Epoch::new(lineage.lineage(), n);
# ring.write(FILE, 0, vec![1; 8], tag(41), ()).expect("a write");
# ring.write(FILE, 8, vec![1; 8], tag(43), ()).expect("a write");
# ring.fail_next_flush(FILE, 1117);
# ring.make_durable_through(tag(41)).expect("a seal");
# let Entry::Failed(f1) = pop_until(&mut ring, |e| matches!(e, Entry::Failed(_))) else {
#     unreachable!()
# };
# let r = ring.write(FILE, 16, vec![1; 8], tag(43), ()).expect("a write");
# pop_until(&mut ring, |e| matches!(e, Entry::Op(done) if done.id == r));
# ring.import_failure(ImportScope::All).expect("the whole instance");
# let Entry::Failed(f2) = pop_until(&mut ring, |e| matches!(e, Entry::Failed(_))) else {
#     unreachable!()
# };
// The trace above, to F2's arrival, is hidden. Then:
ring.resolve(vec![(f1.token, Resolution::Heal), (f2.token, Resolution::Abandon)])
    .expect("heal F1, abandon F2");
// <- Abandoned { failure: F2, suspect: [A@41, P@43, R@43] }
pop_until(&mut ring, |e| matches!(e, Entry::Abandoned { .. }));
assert_eq!(ring.epoch_state(epoch(41)), Ok(EpochState::Abandoned));
assert_eq!(ring.epoch_state(epoch(43)), Ok(EpochState::Abandoned));
// -> make_durable_through(44)                 // F1's heal takes effect here
ring.make_durable_through(tag(44)).expect("a seal");
// <- Durable { through: 44 }                  // "durable through 44" means durable or abandoned
pop_until(&mut ring, |e| matches!(e, Entry::Durable { through } if *through == epoch(44)));
assert_eq!(ring.durable_through(lineage.lineage()), Ok(Some(44)));
assert_eq!(ring.epoch_state(epoch(43)), Ok(EpochState::Abandoned));
# drop(ring);
# std::fs::remove_file(&path)?;
# Ok::<(), std::io::Error>(())