liminal-server 0.19.1

//! A write the store applied but answered `OutcomeUnknown` for.
//!
//! haematite answers a write whose caller stopped waiting after the shard had
//! started it with `OutcomeUnknown`: it may have been applied, so the position
//! must be read again. These pins drive one write of each class through
//! [`UnknownOutcomeStore`], which applies the write and then answers that, and
//! assert what the design table
//! (`docs/design/durability/OUTCOME-UNKNOWN-WRITE-SITES-2026-10-07.md`) says
//! must hold: the live owner equals what the store holds, the next commit
//! succeeds, and the client is answered the committed value, never a failure
//! for a row that is durable.

use std::error::Error;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use liminal::durability::{
    DurabilityError, DurableStore, ScanCursor, ScanPage, StoredEntry, bridge::block_on,
};
use liminal_protocol::wire::{
    AttachAttemptToken, ClientRequest, ConnectionIncarnation, CredentialAttachRequest,
    DetachAttemptToken, DetachRequest, EnrollBound, EnrollmentRequest, EnrollmentToken, Generation,
    LeaveAttemptToken, LeaveRequest, RecordAdmission, RecordAdmissionAttemptToken, ServerValue,
};

use super::ProductionParticipantHandler;
use super::log::{OperationLog, STREAM_PREFIX};
use super::observer::OBSERVER_STREAM_KEY;
use super::outbox_log::OUTBOX_STREAM_PREFIX;
use super::registry::REGISTRY_STREAM_KEY;
use super::tests::{decode_server_value, dispatch, dispatch_outcome, test_participant_config};
use super::tests_memory_store::MemoryStore;
use crate::server::participant::{ParticipantConnectionConversations, ParticipantDispatch};

/// Which kind of write the store answers `OutcomeUnknown` for.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum Write {
    Append,
    AppendAll,
    Cas,
}

/// Applies every write, and answers the first armed write of one kind to keys
/// under one prefix with `OutcomeUnknown` once it has been applied. Armed
/// unsettled, the first read under the prefix after that answer fails too, so
/// the write cannot be settled by reading it back.
#[derive(Debug)]
pub(super) struct UnknownOutcomeStore {
    inner: Arc<dyn DurableStore>,
    write: Write,
    prefix: String,
    armed: AtomicBool,
    blind_on_strike: AtomicBool,
    blind: AtomicBool,
}

impl UnknownOutcomeStore {
    pub(super) fn new(inner: &Arc<dyn DurableStore>, write: Write, prefix: &str) -> Self {
        Self {
            inner: Arc::clone(inner),
            write,
            prefix: prefix.to_owned(),
            armed: AtomicBool::new(false),
            blind_on_strike: AtomicBool::new(false),
            blind: AtomicBool::new(false),
        }
    }

    pub(super) fn arm(&self) {
        self.armed.store(true, Ordering::SeqCst);
    }

    /// Arms the write, and the read after it to fail.
    pub(super) fn arm_unsettled(&self) {
        self.blind_on_strike.store(true, Ordering::SeqCst);
        self.arm();
    }

    /// Whether this write is the armed one, disarming it.
    fn strikes(&self, write: Write, key: &str) -> bool {
        let struck = write == self.write
            && key.starts_with(&self.prefix)
            && self.armed.swap(false, Ordering::SeqCst);
        if struck && self.blind_on_strike.swap(false, Ordering::SeqCst) {
            self.blind.store(true, Ordering::SeqCst);
        }
        struck
    }

    /// Whether this read is the one after the answer that fails, clearing it.
    fn blinded(&self, key: &str) -> bool {
        key.starts_with(&self.prefix) && self.blind.swap(false, Ordering::SeqCst)
    }

    /// Whether the armed write has been answered.
    pub(super) fn struck(&self) -> bool {
        !self.armed.load(Ordering::SeqCst)
    }
}

/// What the store answers a started write whose caller stopped waiting, as
/// it maps haematite's error.
fn outcome_unknown() -> DurabilityError {
    haematite::ApiError::Storage(haematite::DatabaseError::OutcomeUnknown { pid: 1 }).into()
}

/// What the store answers the read that cannot settle the write.
fn unreadable() -> DurabilityError {
    haematite::ApiError::Storage(haematite::DatabaseError::IoError(std::io::Error::other(
        "the read-back of an unknown outcome failed",
    )))
    .into()
}

#[async_trait::async_trait]
impl DurableStore for UnknownOutcomeStore {
    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        let applied = self.inner.append(stream_key, payload, expected_seq).await?;
        if self.strikes(Write::Append, stream_key) {
            return Err(outcome_unknown());
        }
        Ok(applied)
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        let applied = self
            .inner
            .append_all(stream_key, payloads, expected_seq)
            .await?;
        if self.strikes(Write::AppendAll, stream_key) {
            return Err(outcome_unknown());
        }
        Ok(applied)
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        if self.blinded(stream_key) {
            return Err(unreadable());
        }
        self.inner.read_from(stream_key, offset, limit).await
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        if self.blinded(stream_key) {
            return Err(unreadable());
        }
        self.inner.read_at(stream_key, sequence).await
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        self.inner.cas(key, old_value, new_value).await?;
        if self.strikes(Write::Cas, key) {
            return Err(outcome_unknown());
        }
        Ok(())
    }

    async fn read_stream_next_seq(&self, stream_key: &str) -> Result<Option<u64>, DurabilityError> {
        self.inner.read_stream_next_seq(stream_key).await
    }

    async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
        self.inner.read_value(key).await
    }

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        self.inner.scan_page(prefix, cursor, limit).await
    }

    async fn flush(&self) -> Result<(), DurabilityError> {
        self.inner.flush().await
    }
}

const FIRST: u64 = 0x0E_01;
const SECOND: u64 = 0x0E_02;
const SENDER: ConnectionIncarnation = ConnectionIncarnation::new(0x0E, 1);
const RECIPIENT: ConnectionIncarnation = ConnectionIncarnation::new(0x0E, 2);
const LATER: ConnectionIncarnation = ConnectionIncarnation::new(0x0E, 3);

/// The memory store beneath, the fault store over it, and a handler on the
/// fault store.
type Faulty = (
    Arc<dyn DurableStore>,
    Arc<UnknownOutcomeStore>,
    ProductionParticipantHandler,
);

/// A handler over a memory store answering `write` to keys under `prefix`
/// with `OutcomeUnknown` once armed.
fn faulty(write: Write, prefix: &str) -> Result<Faulty, Box<dyn Error>> {
    let memory: Arc<dyn DurableStore> = Arc::new(MemoryStore::default());
    let fault = Arc::new(UnknownOutcomeStore::new(&memory, write, prefix));
    let store: Arc<dyn DurableStore> = fault.clone();
    let handler = ProductionParticipantHandler::new(store, test_participant_config())?;
    Ok((memory, fault, handler))
}

fn enrollment(conversation_id: u64, token: u8) -> ClientRequest {
    ClientRequest::Enrollment(EnrollmentRequest {
        conversation_id,
        enrollment_token: EnrollmentToken::new([token; 16]),
    })
}

fn enrolled(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    conversation_id: u64,
    token: u8,
) -> Result<u64, Box<dyn Error>> {
    let value = dispatch(handler, connection, enrollment(conversation_id, token))?;
    let ServerValue::EnrollBound(bound) = value else {
        return Err(format!("enrollment {token:#x} did not bind: {value:?}").into());
    };
    Ok(bound.participant_id())
}

fn record(conversation_id: u64, sender: u64, index: u8) -> ClientRequest {
    ClientRequest::RecordAdmission(RecordAdmission {
        conversation_id,
        participant_id: sender,
        capability_generation: Generation::ONE,
        record_admission_attempt_token: RecordAdmissionAttemptToken::new([index; 16]),
        payload: vec![index],
    })
}

/// The operation rows the store holds for `conversation_id`.
fn stored_rows(
    memory: &Arc<dyn DurableStore>,
    conversation_id: u64,
) -> Result<u64, Box<dyn Error>> {
    let log = OperationLog::new(Arc::clone(memory), conversation_id);
    let mut rows = 0_u64;
    while block_on(log.read_at(rows))??.is_some() {
        rows += 1;
    }
    Ok(rows)
}

/// The live owner's next log sequence for `conversation_id`.
fn live_rows(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
) -> Result<Option<u64>, Box<dyn Error>> {
    let cell = handler.cell(conversation_id)?;
    let owner = cell.lock().map_err(|_| "the owner lock was poisoned")?;
    let rows = owner.as_ref().map(|authority| authority.next_log_sequence);
    drop(owner);
    Ok(rows)
}

/// A sender and a recipient enrolled in `conversation_id`; answers the sender.
fn two_members(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
) -> Result<u64, Box<dyn Error>> {
    let sender = enrolled(handler, SENDER, conversation_id, 0xE1)?;
    enrolled(handler, RECIPIENT, conversation_id, 0xE2)?;
    Ok(sender)
}

/// Class 1: a record whose operation row landed under an unknown outcome is
/// answered its committed value, the owner matches the store, and the next
/// record commits.
#[test]
fn a_record_whose_row_landed_under_an_unknown_outcome_is_answered_committed()
-> Result<(), Box<dyn Error>> {
    let (memory, fault, handler) = faulty(Write::Append, &format!("{STREAM_PREFIX}{FIRST}"))?;
    let sender = two_members(&handler, FIRST)?;
    fault.arm();
    let answered = dispatch(&handler, SENDER, record(FIRST, sender, 0x01));
    assert!(fault.struck(), "the record wrote no operation row");
    let next = dispatch(&handler, SENDER, record(FIRST, sender, 0x02));
    let stored = stored_rows(&memory, FIRST)?;
    let live = live_rows(&handler, FIRST)?;
    let answered = answered.map_err(|error| error.to_string());
    assert!(
        matches!(answered, Ok(ServerValue::RecordCommitted(_))),
        "a record whose row the store holds was answered {answered:?}, not its committed value"
    );
    assert!(
        matches!(next, Ok(ServerValue::RecordCommitted(_))),
        "the next record did not commit"
    );
    assert_eq!(
        live,
        Some(stored),
        "the live owner disagrees with the store"
    );
    Ok(())
}

/// Class 2: a record whose extension row landed under an unknown outcome is
/// answered its committed value, and the next record commits.
#[test]
fn a_record_whose_extension_row_landed_under_an_unknown_outcome_is_answered_committed()
-> Result<(), Box<dyn Error>> {
    let (memory, fault, handler) =
        faulty(Write::Append, &format!("{OUTBOX_STREAM_PREFIX}{FIRST}"))?;
    let sender = two_members(&handler, FIRST)?;
    fault.arm();
    let answered = dispatch(&handler, SENDER, record(FIRST, sender, 0x01));
    assert!(fault.struck(), "the record wrote no extension row");
    let next = dispatch(&handler, SENDER, record(FIRST, sender, 0x02));
    let stored = stored_rows(&memory, FIRST)?;
    let live = live_rows(&handler, FIRST)?;
    let answered = answered.map_err(|error| error.to_string());
    assert!(
        matches!(answered, Ok(ServerValue::RecordCommitted(_))),
        "a record whose extension row the store holds was answered {answered:?}, not its \
         committed value"
    );
    assert!(
        matches!(next, Ok(ServerValue::RecordCommitted(_))),
        "the next record did not commit"
    );
    assert_eq!(
        live,
        Some(stored),
        "the live owner disagrees with the store"
    );
    Ok(())
}

/// Class 3: a registry row that landed under an unknown outcome leaves the
/// next conversation creatable, and the first conversation's enrollment is
/// answered as the store settles it.
#[test]
fn a_registry_row_landed_under_an_unknown_outcome_leaves_the_next_conversation_creatable()
-> Result<(), Box<dyn Error>> {
    let (_memory, fault, handler) = faulty(Write::Append, REGISTRY_STREAM_KEY)?;
    fault.arm();
    let first = dispatch(&handler, SENDER, enrollment(FIRST, 0xE1));
    assert!(fault.struck(), "the first enrollment wrote no registry row");
    let second = dispatch(&handler, LATER, enrollment(SECOND, 0xE3));
    let first = first.map_err(|error| error.to_string());
    let second = second.map_err(|error| error.to_string());
    assert!(
        matches!(second, Ok(ServerValue::EnrollBound(_))),
        "after a registry row landed under an unknown outcome, the next conversation's creation \
         was answered {second:?}"
    );
    assert!(
        matches!(first, Ok(ServerValue::EnrollBound(_))),
        "the enrollment whose registry row the store holds was answered {first:?}"
    );
    Ok(())
}

/// Class 4: an observer row that landed under an unknown outcome leaves the
/// observer owner in step with its log: the enrollment that wrote it is
/// answered as the store settles it, and the next conversation is tracked.
#[test]
fn an_observer_row_landed_under_an_unknown_outcome_leaves_the_next_conversation_tracked()
-> Result<(), Box<dyn Error>> {
    let (_memory, fault, handler) = faulty(Write::AppendAll, OBSERVER_STREAM_KEY)?;
    fault.arm();
    let first = dispatch(&handler, SENDER, enrollment(FIRST, 0xE1));
    assert!(fault.struck(), "the first enrollment wrote no observer row");
    let second = dispatch(&handler, LATER, enrollment(SECOND, 0xE3));
    let first = first.map_err(|error| error.to_string());
    let second = second.map_err(|error| error.to_string());
    assert!(
        matches!(second, Ok(ServerValue::EnrollBound(_))),
        "after an observer row landed under an unknown outcome, the next conversation's \
         enrollment was answered {second:?}"
    );
    assert!(
        matches!(first, Ok(ServerValue::EnrollBound(_))),
        "the enrollment whose observer row the store holds was answered {first:?}"
    );
    Ok(())
}

/// What the client is told for one dispatch: the frame's discriminant, and
/// whether the connection closes after it.
fn told(outcome: &ParticipantDispatch) -> Result<String, Box<dyn Error>> {
    let (frame, closes) = match outcome {
        ParticipantDispatch::Respond(frame) => (frame, false),
        ParticipantDispatch::RespondThenClose(frame) => (frame, true),
        ParticipantDispatch::Fatal(_) => return Ok("no frame, then close".to_owned()),
        ParticipantDispatch::NotParticipant => return Ok("not a participant frame".to_owned()),
    };
    let mut bytes = vec![0; liminal::protocol::encoded_len(frame)?];
    let written = liminal::protocol::encode(frame, &mut bytes)?;
    bytes.truncate(written);
    let discriminant: [u8; 2] = bytes
        .get(14..16)
        .and_then(|discriminant| discriminant.try_into().ok())
        .ok_or("the response frame has no participant discriminant")?;
    let discriminant = u16::from_be_bytes(discriminant);
    Ok(if closes {
        format!("frame {discriminant:#06x}, then close")
    } else {
        format!("frame {discriminant:#06x}")
    })
}

/// Class 1 at the wire, when the store cannot say: a record whose row landed
/// under an unknown outcome that its read-back cannot settle is answered by
/// the named outcome-unknown frame (0x0128) on a connection that stays open,
/// never a bare close; its retry with the same token is answered its commit,
/// and the row is stored once.
#[test]
fn an_unsettled_record_is_answered_outcome_unknown_and_its_retry_its_commit()
-> Result<(), Box<dyn Error>> {
    let (memory, fault, handler) = faulty(Write::Append, &format!("{STREAM_PREFIX}{FIRST}"))?;
    let sender = two_members(&handler, FIRST)?;
    let before = stored_rows(&memory, FIRST)?;
    fault.arm_unsettled();
    let mut conversations = ParticipantConnectionConversations::default();
    let answered = dispatch_outcome(
        &handler,
        SENDER,
        &mut conversations,
        record(FIRST, sender, 0x01),
    )?;
    assert!(fault.struck(), "the record wrote no operation row");
    let answered = told(&answered)?;
    let retried = dispatch_outcome(
        &handler,
        SENDER,
        &mut conversations,
        record(FIRST, sender, 0x01),
    )?;
    let retried = match retried {
        ParticipantDispatch::Respond(frame) => match decode_server_value(&frame)? {
            ServerValue::RecordCommitted(committed) => {
                format!("committed at {:?}", committed.delivery_seq())
            }
            other => format!("{other:?}"),
        },
        other => told(&other)?,
    };
    let stored = stored_rows(&memory, FIRST)? - before;
    assert_eq!(
        (answered.as_str(), stored),
        ("frame 0x0128", 1),
        "a record the store could not settle was not answered by the outcome-unknown frame, or was stored other than once"
    );
    assert!(
        retried.starts_with("committed at"),
        "the same-token retry of an unsettled record was answered {retried}, not its commit"
    );
    Ok(())
}

/// Answers `request` under an unsettled outcome, then retries it with the
/// same token on the same connection: what the client is told first, the
/// retry's answer, and the operation rows the two wrote, counted on the store
/// beneath the fault.
fn unsettled_then_retried(
    handler: &ProductionParticipantHandler,
    (memory, fault): (&Arc<dyn DurableStore>, &UnknownOutcomeStore),
    connection: ConnectionIncarnation,
    request: &ClientRequest,
) -> Result<(String, String, u64), Box<dyn Error>> {
    let before = stored_rows(memory, FIRST)?;
    let mut conversations = ParticipantConnectionConversations::default();
    fault.arm_unsettled();
    let first = dispatch_outcome(handler, connection, &mut conversations, request.clone())?;
    if !fault.struck() {
        return Err("the request wrote no operation row".into());
    }
    let first = told(&first)?;
    let retried = dispatch_outcome(handler, connection, &mut conversations, request.clone())?;
    let retried = match retried {
        ParticipantDispatch::Respond(frame) => {
            format!("{:?}", decode_server_value(&frame)?.discriminant())
        }
        other => told(&other)?,
    };
    Ok((first, retried, stored_rows(memory, FIRST)? - before))
}

/// A bound member of `FIRST` on `connection`.
fn bound(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    token: u8,
) -> Result<EnrollBound, Box<dyn Error>> {
    let value = dispatch(handler, connection, enrollment(FIRST, token))?;
    let ServerValue::EnrollBound(bound) = value else {
        return Err(format!("enrollment {token:#x} did not bind: {value:?}").into());
    };
    Ok(bound)
}

fn detach_request(member: &EnrollBound, token: u8) -> ClientRequest {
    ClientRequest::Detach(DetachRequest {
        conversation_id: FIRST,
        participant_id: member.participant_id(),
        capability_generation: member.capability_generation(),
        detach_attempt_token: DetachAttemptToken::new([token; 16]),
    })
}

/// Each token-carrying operation whose operation row landed under an outcome
/// its read-back cannot settle is told the outcome-unknown frame on an open
/// connection, and the same request with the same token, on that connection,
/// is answered the operation's original commit: a binding by its receipt's
/// replay (`Bound`), a detach or leave by its commit. Either way the
/// operation's row is stored once: the retry writes none.
#[test]
fn an_unsettled_enrollment_is_answered_outcome_unknown_and_its_retry_its_binding()
-> Result<(), Box<dyn Error>> {
    let (memory, fault, handler) = faulty(Write::Append, &format!("{STREAM_PREFIX}{FIRST}"))?;
    bound(&handler, SENDER, 0xE1)?;
    let answered = unsettled_then_retried(
        &handler,
        (&memory, &fault),
        RECIPIENT,
        &enrollment(FIRST, 0xE2),
    )?;
    assert_eq!(
        answered,
        ("frame 0x0128".to_owned(), "Bound".to_owned(), 1),
        "an unsettled enrollment was not told outcome unknown, then bound on its retry, with its \
         operation stored once"
    );
    Ok(())
}

#[test]
fn an_unsettled_credential_attach_is_answered_outcome_unknown_and_its_retry_its_binding()
-> Result<(), Box<dyn Error>> {
    let (memory, fault, handler) = faulty(Write::Append, &format!("{STREAM_PREFIX}{FIRST}"))?;
    let member = bound(&handler, SENDER, 0xE1)?;
    let detached = dispatch(&handler, SENDER, detach_request(&member, 0xD1))?;
    if !matches!(detached, ServerValue::DetachCommitted(_)) {
        return Err(format!("the member did not detach: {detached:?}").into());
    }
    let attach = ClientRequest::CredentialAttach(CredentialAttachRequest {
        conversation_id: FIRST,
        participant_id: member.participant_id(),
        capability_generation: member.capability_generation(),
        attach_secret: member.attach_secret(),
        attach_attempt_token: AttachAttemptToken::new([0xA1; 16]),
        accept_marker_delivery_seq: None,
    });
    let answered = unsettled_then_retried(&handler, (&memory, &fault), LATER, &attach)?;
    assert_eq!(
        answered,
        ("frame 0x0128".to_owned(), "Bound".to_owned(), 1),
        "an unsettled credential attach was not told outcome unknown, then bound on its retry, with its \
         operation stored once"
    );
    Ok(())
}

#[test]
fn an_unsettled_detach_is_answered_outcome_unknown_and_its_retry_its_commit()
-> Result<(), Box<dyn Error>> {
    let (memory, fault, handler) = faulty(Write::Append, &format!("{STREAM_PREFIX}{FIRST}"))?;
    let member = bound(&handler, SENDER, 0xE1)?;
    let answered = unsettled_then_retried(
        &handler,
        (&memory, &fault),
        SENDER,
        &detach_request(&member, 0xD1),
    )?;
    assert_eq!(
        answered,
        ("frame 0x0128".to_owned(), "DetachCommitted".to_owned(), 1),
        "an unsettled detach was not told outcome unknown, then committed on its retry, with its \
         operation stored once"
    );
    Ok(())
}

#[test]
fn an_unsettled_leave_is_answered_outcome_unknown_and_its_retry_its_commit()
-> Result<(), Box<dyn Error>> {
    let (memory, fault, handler) = faulty(Write::Append, &format!("{STREAM_PREFIX}{FIRST}"))?;
    let member = bound(&handler, SENDER, 0xE1)?;
    let leave = ClientRequest::Leave(LeaveRequest {
        conversation_id: FIRST,
        participant_id: member.participant_id(),
        capability_generation: member.capability_generation(),
        attach_secret: member.attach_secret(),
        leave_attempt_token: LeaveAttemptToken::new([0x1E; 16]),
    });
    let answered = unsettled_then_retried(&handler, (&memory, &fault), SENDER, &leave)?;
    assert_eq!(
        answered,
        ("frame 0x0128".to_owned(), "LeaveCommitted".to_owned(), 1),
        "an unsettled leave was not told outcome unknown, then committed on its retry, with its \
         operation stored once"
    );
    Ok(())
}