ursula-runtime 0.5.0

use std::collections::HashSet;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
use std::time::Duration;

use bytes::Bytes;
use serde_json::json;
use tokio::sync::Notify;
use tokio::sync::Semaphore;
use tokio::sync::oneshot;
use ursula_shard::BucketStreamId;
use ursula_shard::CoreId;
use ursula_shard::RaftGroupId;
use ursula_shard::ShardId;
use ursula_shard::ShardPlacement;
use ursula_stream::ExternalPayloadRef;
use ursula_stream::ObjectPayloadRef;
use ursula_stream::StreamAttrs;
use ursula_stream::StreamCommand;
use ursula_stream::StreamReadColdIndexSegment;
use ursula_stream::StreamReadPlan;
use ursula_stream::StreamReadSegment;
use ursula_stream::StreamSnapshot;
use ursula_stream::StreamStateMachine;

use super::*;
use crate::ColdReadCacheParams;
use crate::cold_store::ColdStoreEvent;
use crate::cold_store::DEFAULT_CONTENT_TYPE;
use crate::core_worker::CoreWorker;
use crate::core_worker::ReadWatcher;
use crate::core_worker::ReadWatchers;
use crate::error::ErrorStatus;
use crate::metrics::RuntimeMetricsInner;

fn runtime(core_count: usize, group_count: usize) -> ShardRuntime {
    ShardRuntime::spawn(test_config(core_count, group_count, 128)).expect("spawn runtime")
}

/// The shared hosted-tokio test config; only the knobs that vary across
/// tests are parameters.
fn test_config(
    core_count: usize,
    raft_group_count: usize,
    mailbox_capacity: usize,
) -> RuntimeConfig {
    RuntimeConfig {
        core_count,
        raft_group_count,
        mailbox_capacity,
        threading: RuntimeThreading::HostedTokio,
        cold_max_hot_bytes_per_group: None,
        raft_max_uncommitted_bytes_per_group: None,
        live_read_max_waiters_per_core: Some(65_536),
    }
}

async fn append_bytes(runtime: &ShardRuntime, stream: &BucketStreamId, payload: &[u8]) {
    runtime
        .append(AppendRequest::from_bytes(stream.clone(), payload.to_vec()))
        .await
        .expect("append");
}

fn memory_cold_store() -> ColdStore {
    ColdStore::memory().expect("memory cold store")
}

fn read_req(stream_id: BucketStreamId, offset: u64, max_len: usize) -> ReadStreamRequest {
    ReadStreamRequest {
        stream_id,
        offset,
        max_len,
        now_ms: 0,
        record: None,
        max_records: None,
        leader_only: false,
    }
}

fn spawn_with_cold_store(config: RuntimeConfig, cold_store: Arc<ColdStore>) -> ShardRuntime {
    ShardRuntime::spawn_with_engine_factory_and_cold_store(
        config,
        InMemoryGroupEngineFactory::with_cold_store(Some(cold_store.clone())),
        Some(cold_store),
    )
    .expect("spawn runtime")
}

fn spawn_in_memory(config: RuntimeConfig) -> ShardRuntime {
    ShardRuntime::spawn_with_engine_factory(config, InMemoryGroupEngineFactory::default())
        .expect("spawn runtime")
}

fn stream_on_group(runtime: &ShardRuntime, group_id: RaftGroupId, prefix: &str) -> BucketStreamId {
    stream_in_bucket_on_group(runtime, group_id, "benchcmp", prefix)
}

fn stream_in_bucket_on_group(
    runtime: &ShardRuntime,
    group_id: RaftGroupId,
    bucket_id: &str,
    prefix: &str,
) -> BucketStreamId {
    for index in 0..10_000 {
        let stream = BucketStreamId::new(bucket_id, format!("{prefix}-{index}"));
        if runtime.locate(&stream).raft_group_id == group_id {
            return stream;
        }
    }
    panic!("could not find stream for group {}", group_id.0);
}

async fn create_stream(runtime: &ShardRuntime, stream: &BucketStreamId) -> CreateStreamResponse {
    runtime
        .create_stream(CreateStreamRequest::new(
            stream.clone(),
            DEFAULT_CONTENT_TYPE,
        ))
        .await
        .expect("create stream")
}

fn producer(id: &str, epoch: u64, seq: u64) -> ProducerRequest {
    ProducerRequest {
        producer_id: id.to_owned(),
        producer_epoch: epoch,
        producer_seq: seq,
    }
}

fn empty_integrity() -> StreamIntegritySnapshot {
    let empty = "00".repeat(32);
    StreamIntegritySnapshot {
        live_setsum: empty.clone(),
        evicted_setsum: empty.clone(),
        total_setsum: empty,
        live_start_offset: 0,
        tail_offset: 0,
        live_records: 0,
        evicted_records: 0,
        total_records: 0,
    }
}

fn stream_attrs(title: &str, purpose: &str) -> StreamAttrs {
    StreamAttrs {
        title: Some(title.to_owned()),
        metadata: json!({
            "agent": { "id": "agent-1", "version": 2 },
            "purpose": purpose
        })
        .as_object()
        .expect("metadata object")
        .clone(),
    }
}

#[test]
fn group_write_command_decodes_pre_attrs_records() {
    let mut request = CreateStreamRequest::new(
        BucketStreamId::new("benchcmp", "legacy-record"),
        "application/octet-stream",
    );
    request.now_ms = 7;
    let command = GroupWriteCommand::from(request);

    let mut value = serde_json::to_value(&command).expect("encode command");
    let fields = value
        .pointer_mut("/Stream/CreateStream")
        .and_then(serde_json::Value::as_object_mut)
        .expect("command object");
    assert!(fields.remove("attrs").is_some());

    let decoded: GroupWriteCommand =
        serde_json::from_value(value).expect("decode pre-attrs record");
    assert_eq!(decoded, command);
}

#[test]
fn stream_from_replicated_preserves_wire_message() {
    let err = GroupEngineError::stream_from_replicated(
        "wire message already includes code",
        StreamErrorCode::StreamGone,
        Some(7),
        vec![StreamErrorContext::StreamClosed],
    );

    assert_eq!(err.message(), "wire message already includes code");
    assert_eq!(err.code(), Some(StreamErrorCode::StreamGone));
    assert_eq!(err.next_offset(), Some(7));
    assert_eq!(err.context(), &[StreamErrorContext::StreamClosed]);
}

#[test]
fn stream_parts_returns_stream_fields_only() {
    let err = GroupEngineError::stream_from_replicated(
        "wire stream message",
        StreamErrorCode::ProducerSeqConflict,
        Some(11),
        vec![StreamErrorContext::ProducerSeqConflict {
            expected_seq: 10,
            received_seq: 7,
        }],
    );

    let (message, code, next_offset, context) = err
        .stream_parts()
        .expect("stream error exposes stream parts");
    assert_eq!(message, "wire stream message");
    assert_eq!(code, StreamErrorCode::ProducerSeqConflict);
    assert_eq!(next_offset, Some(11));
    assert_eq!(context, &[StreamErrorContext::ProducerSeqConflict {
        expected_seq: 10,
        received_seq: 7,
    }]);

    assert!(GroupEngineError::new("infra").stream_parts().is_none());
    assert!(
        GroupEngineError::forward_to_leader("forward", None, None)
            .stream_parts()
            .is_none()
    );
}

#[test]
fn runtime_error_status_classifies_retryable_and_permanent_errors() {
    let backpressure = RuntimeError::LiveReadBackpressure {
        core_id: CoreId(0),
        current_waiters: 65_536,
        limit: 65_536,
    };
    assert_eq!(backpressure.status(), ErrorStatus::Temporary);

    let conflict = RuntimeError::GroupEngine {
        core_id: CoreId(0),
        raft_group_id: RaftGroupId(0),
        error: GroupEngineError::stream(
            StreamErrorCode::StreamSeqConflict,
            "expected sequence 1 received 0",
        ),
    };
    assert_eq!(conflict.status(), ErrorStatus::Permanent);

    let internal = RuntimeError::GroupEngine {
        core_id: CoreId(0),
        raft_group_id: RaftGroupId(0),
        error: GroupEngineError::new("OpenRaft client_write: timeout after retries"),
    };
    assert_eq!(internal.status(), ErrorStatus::Persistent);
}

#[test]
fn group_engine_error_variants_separate_stream_infra_and_forwarding() {
    let stream = GroupEngineError::stream_with_context(
        StreamErrorCode::ProducerSeqConflict,
        "producer conflict",
        Some(9),
        vec![StreamErrorContext::ProducerSeqConflict {
            expected_seq: 8,
            received_seq: 3,
        }],
    );
    assert!(matches!(stream, GroupEngineError::Stream(_)));

    let infra = GroupEngineError::new("OpenRaft client_write failed");
    assert!(matches!(
        infra,
        GroupEngineError::Infra(GroupInfraError::Internal { .. })
    ));

    let forward = GroupEngineError::forward_to_leader("forward to leader", Some(2), None);
    assert!(matches!(forward, GroupEngineError::ForwardToLeader { .. }));
}

#[test]
fn stale_cold_flush_candidate_classification_uses_context_not_message_text() {
    let placement = ShardPlacement {
        core_id: CoreId(0),
        shard_id: ShardId(0),
        raft_group_id: RaftGroupId(0),
    };
    let message = "cold chunk end 18 is beyond stream 'benchcmp/stale' tail 17";
    let without_context = RuntimeError::group_engine(
        placement,
        GroupEngineError::stream(StreamErrorCode::InvalidColdFlush, message),
    );
    assert!(
        !crate::metrics::is_stale_cold_flush_candidate_error(&without_context),
        "stale classification must not be inferred from message text"
    );

    let with_context = RuntimeError::group_engine(
        placement,
        GroupEngineError::stream_with_context(
            StreamErrorCode::InvalidColdFlush,
            "cold flush candidate is stale",
            None,
            vec![StreamErrorContext::StaleColdFlushCandidate],
        ),
    );
    assert!(crate::metrics::is_stale_cold_flush_candidate_error(
        &with_context
    ));
}

fn placement() -> ShardPlacement {
    ShardPlacement {
        core_id: CoreId(0),
        shard_id: ShardId(0),
        raft_group_id: RaftGroupId(0),
    }
}

#[test]
fn group_write_command_round_trips_as_log_payload() {
    let command = GroupWriteCommand::Stream(StreamCommand::AppendBatch {
        stream_id: BucketStreamId::new("benchcmp", "raft-log"),
        content_type: Some(DEFAULT_CONTENT_TYPE.to_owned()),
        payloads: vec![Bytes::from_static(b"ab"), Bytes::from_static(b"cd")],
        producer: Some(producer("writer-1", 7, 42)),
        now_ms: 0,
    });

    let encoded = serde_json::to_vec(&command).expect("encode command");
    let decoded = serde_json::from_slice::<GroupWriteCommand>(&encoded).expect("decode command");

    assert_eq!(decoded, command);
}

#[test]
fn committed_write_command_is_state_machine_apply_boundary() {
    let placement = ShardPlacement {
        core_id: CoreId(0),
        shard_id: ShardId(0),
        raft_group_id: RaftGroupId(0),
    };
    let stream = BucketStreamId::new("benchcmp", "apply-command");
    let mut engine = InMemoryGroupEngine::default();

    let created = engine
        .apply_committed_write(
            GroupWriteCommand::Stream(StreamCommand::CreateStream {
                stream_id: stream.clone(),
                content_type: DEFAULT_CONTENT_TYPE.to_owned(),
                initial_payload: Bytes::new(),
                close_after: false,
                stream_seq: None,
                producer: None,
                stream_ttl_seconds: None,
                stream_expires_at_ms: None,
                attrs: None,
                now_ms: 0,
            }),
            placement,
        )
        .expect("create stream");
    assert_eq!(
        created,
        GroupWriteResponse::CreateStream(CreateStreamResponse {
            placement,
            next_offset: 0,
            closed: false,
            already_exists: false,
            group_commit_index: 1,
            record_range: None,
        })
    );

    let appended = engine
        .apply_committed_write(
            GroupWriteCommand::Stream(StreamCommand::Append {
                stream_id: stream.clone(),
                content_type: Some(DEFAULT_CONTENT_TYPE.to_owned()),
                payload: Bytes::from_static(b"abc"),
                close_after: false,
                stream_seq: None,
                producer: None,
                now_ms: 0,
                record_match: None,
            }),
            placement,
        )
        .expect("append");
    assert_eq!(
        appended,
        GroupWriteResponse::Append(AppendResponse {
            placement,
            start_offset: 0,
            next_offset: 3,
            stream_append_count: 1,
            group_commit_index: 2,
            closed: false,
            deduplicated: false,
            producer: None,
            record_range: None,
            stream_hot_bytes: 3,
            group_hot_bytes: 3,
        })
    );

    let flushed = engine
        .apply_committed_write(
            GroupWriteCommand::Stream(StreamCommand::FlushCold {
                stream_id: stream.clone(),
                chunk: ColdChunkRef {
                    start_offset: 0,
                    end_offset: 2,
                    s3_path: "s3://bucket/apply-command/000000".to_owned(),
                    object_size: 2,
                    ..Default::default()
                },
            }),
            placement,
        )
        .expect("flush cold");
    assert_eq!(
        flushed,
        GroupWriteResponse::FlushCold(FlushColdResponse {
            placement,
            hot_start_offset: 2,
            group_commit_index: 3,
        })
    );

    let read = engine
        .state_machine
        .read(&stream, 2, 16)
        .expect("read applied command");
    assert_eq!(read.payload, b"c");
    let plan = engine
        .state_machine
        .read_plan(&stream, 0, 16)
        .expect("read plan");
    assert_eq!(plan.segments.len(), 2);
    assert!(matches!(plan.segments[0], StreamReadSegment::ColdIndex(_)));
    assert_eq!(plan.segments[1], StreamReadSegment::Hot(b"c".to_vec()));
}

#[tokio::test]
async fn cold_store_read_reassembles_cold_and_hot_segments() {
    let placement = placement();
    let stream = BucketStreamId::new("benchcmp", "cold-read");
    let cold_store = Arc::new(memory_cold_store());
    cold_store
        .write_chunk("benchcmp/cold-read/chunks/000000.bin", b"abcd")
        .await
        .expect("write cold object");
    let mut engine = InMemoryGroupEngine::with_cold_store(cold_store);

    engine
        .apply_committed_write(
            GroupWriteCommand::Stream(StreamCommand::CreateStream {
                stream_id: stream.clone(),
                content_type: DEFAULT_CONTENT_TYPE.to_owned(),
                initial_payload: Bytes::new(),
                close_after: false,
                stream_seq: None,
                producer: None,
                stream_ttl_seconds: None,
                stream_expires_at_ms: None,
                attrs: None,
                now_ms: 0,
            }),
            placement,
        )
        .expect("create stream");
    engine
        .apply_committed_write(
            GroupWriteCommand::Stream(StreamCommand::Append {
                stream_id: stream.clone(),
                content_type: Some(DEFAULT_CONTENT_TYPE.to_owned()),
                payload: Bytes::from_static(b"abcdef"),
                close_after: false,
                stream_seq: None,
                producer: None,
                now_ms: 0,
                record_match: None,
            }),
            placement,
        )
        .expect("append");
    engine
        .flush_cold(
            FlushColdRequest {
                stream_id: stream.clone(),
                chunk: ColdChunkRef {
                    start_offset: 0,
                    end_offset: 4,
                    s3_path: "benchcmp/cold-read/chunks/000000.bin".to_owned(),
                    object_size: 4,
                    ..Default::default()
                },
            },
            placement,
        )
        .await
        .expect("flush cold");

    let read = engine
        .read_stream(read_req(stream, 2, 4), placement)
        .await
        .expect("read cold and hot segments");
    assert_eq!(read.payload, b"cdef");
    assert_eq!(read.next_offset, 6);
    assert!(read.up_to_date);
}

#[tokio::test]
async fn cold_index_read_materializes_overlapping_flush_objects_once() {
    let stream = BucketStreamId::new("benchcmp", "overlapping-cold-read");
    let cold_store = Arc::new(memory_cold_store());
    let first = ColdChunkRef {
        start_offset: 0,
        end_offset: 4,
        s3_path: "benchcmp/overlapping-cold-read/chunks/first.bin".to_owned(),
        object_size: 4,
        ..Default::default()
    };
    let second = ColdChunkRef {
        start_offset: 2,
        end_offset: 6,
        s3_path: "benchcmp/overlapping-cold-read/chunks/second.bin".to_owned(),
        object_size: 4,
        ..Default::default()
    };
    cold_store
        .write_chunk(&first.s3_path, b"abcd")
        .await
        .expect("write first cold object");
    cold_store
        .write_chunk(&second.s3_path, b"cdef")
        .await
        .expect("write second cold object");
    let page_store = ColdStoreColdIndexPageStore::new(cold_store.clone());
    write_cold_chunk_index_pages(&page_store, &stream, &first)
        .await
        .expect("index first cold object");
    write_cold_chunk_index_pages(&page_store, &stream, &second)
        .await
        .expect("index overlapping cold object");
    let cache = Arc::new(ColdIndexPageCache::new(Arc::new(page_store), 8));
    let plan = StreamReadPlan {
        offset: 0,
        next_offset: 6,
        content_type: DEFAULT_CONTENT_TYPE.to_owned(),
        segments: vec![StreamReadSegment::ColdIndex(StreamReadColdIndexSegment {
            generation: 0,
            page_id: 0,
            read_start_offset: 0,
            len: 6,
        })],
        up_to_date: true,
        closed: false,
        retained_record_range: None,
        record_range: None,
    };

    let payload = InMemoryGroupEngine::read_payload_from_plan(
        Some(&cold_store),
        Some(&cache),
        &stream,
        &plan,
    )
    .await
    .expect("materialize overlapping cold objects");
    assert_eq!(payload, b"abcdef");
}

#[tokio::test]
async fn stale_cold_flush_rolls_back_index_page_entry() {
    let placement = placement();
    let stream = BucketStreamId::new("benchcmp", "stale-cold-index");
    let live_path = "benchcmp/stale-cold-index/chunks/live.bin";
    let stale_path = "benchcmp/stale-cold-index/chunks/stale.bin";
    let cold_store = Arc::new(memory_cold_store());
    cold_store
        .write_chunk(live_path, b"abcd")
        .await
        .expect("write live cold object");
    cold_store
        .write_chunk(stale_path, b"abcd")
        .await
        .expect("write stale cold object");
    let mut engine = InMemoryGroupEngine::with_cold_store(cold_store.clone());

    engine
        .create_stream(
            CreateStreamRequest::new(stream.clone(), DEFAULT_CONTENT_TYPE),
            placement,
            ColdWriteAdmission::default(),
        )
        .await
        .expect("create stream");
    engine
        .append(
            AppendRequest::from_bytes(stream.clone(), b"abcdef".to_vec()),
            placement,
            ColdWriteAdmission::default(),
        )
        .await
        .expect("append");
    engine
        .flush_cold(
            FlushColdRequest {
                stream_id: stream.clone(),
                chunk: ColdChunkRef {
                    start_offset: 0,
                    end_offset: 4,
                    s3_path: live_path.to_owned(),
                    object_size: 4,
                    ..Default::default()
                },
            },
            placement,
        )
        .await
        .expect("flush live cold chunk");

    let stale_flush = engine
        .flush_cold(
            FlushColdRequest {
                stream_id: stream.clone(),
                chunk: ColdChunkRef {
                    start_offset: 0,
                    end_offset: 4,
                    s3_path: stale_path.to_owned(),
                    object_size: 4,
                    ..Default::default()
                },
            },
            placement,
        )
        .await
        .expect_err("duplicate cold flush should be stale");
    assert!(
        stale_flush
            .message()
            .contains("must start at the hot prefix"),
        "message={}",
        stale_flush.message()
    );
    cold_store
        .delete_chunk(stale_path)
        .await
        .expect("delete stale cold object");

    let read = engine
        .read_stream(read_req(stream, 0, 6), placement)
        .await
        .expect("read should still use live cold index entry");
    assert_eq!(read.payload, b"abcdef");
}

#[tokio::test]
async fn failed_cold_compaction_publish_rolls_back_index_replacement() {
    let placement = placement();
    let stream = BucketStreamId::new("benchcmp", "failed-compact-index");
    let cold_store = Arc::new(memory_cold_store());
    let page_store = ColdStoreColdIndexPageStore::new(cold_store.clone());
    let first = ColdChunkRef {
        start_offset: 0,
        end_offset: 4,
        s3_path: "benchcmp/failed-compact-index/chunks/first.bin".to_owned(),
        object_size: 4,
        ..Default::default()
    };
    let second = ColdChunkRef {
        start_offset: 4,
        end_offset: 8,
        s3_path: "benchcmp/failed-compact-index/chunks/second.bin".to_owned(),
        object_size: 4,
        ..Default::default()
    };
    let replacement = ColdChunkRef {
        start_offset: 0,
        end_offset: 8,
        s3_path: "benchcmp/failed-compact-index/chunks/replacement.bin".to_owned(),
        object_size: 8,
        ..Default::default()
    };
    for chunk in [&first, &second] {
        write_cold_chunk_index_pages(&page_store, &stream, chunk)
            .await
            .expect("index input chunk");
    }
    let mut engine = InMemoryGroupEngine::with_cold_store(cold_store);

    engine
        .compact_cold(
            CompactColdRequest {
                stream_id: stream.clone(),
                old_chunks: vec![first.clone(), second.clone()],
                replacement,
                gc_not_before_ms: 0,
            },
            placement,
        )
        .await
        .expect_err("missing stream must reject compaction publish");

    let chunks = load_cold_chunks_from_pages(&page_store, &[ColdIndexPageKey {
        stream_id: stream,
        generation: 0,
        page_id: 0,
    }])
    .await
    .expect("load rolled-back index");
    assert_eq!(chunks, vec![first, second]);
}

#[tokio::test]
async fn external_payload_index_pages_are_not_kept_in_snapshot_memory() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(RuntimeConfig::new(1, 1), cold_store.clone());
    let stream = BucketStreamId::new("benchcmp", "external-index");

    cold_store
        .write_chunk("benchcmp/external-index/external/initial.bin", b"ab")
        .await
        .expect("write initial external payload");
    runtime
        .create_stream_external(CreateStreamExternalRequest {
            stream_id: stream.clone(),
            content_type: DEFAULT_CONTENT_TYPE.to_owned(),
            initial_payload: ExternalPayloadRef {
                s3_path: "benchcmp/external-index/external/initial.bin".to_owned(),
                payload_len: 2,
                object_size: 2,
            },
            record_ends: Vec::new(),
            close_after: false,
            stream_seq: None,
            producer: None,
            stream_ttl_seconds: None,
            stream_expires_at_ms: None,
            attrs: None,
            now_ms: 0,
        })
        .await
        .expect("create external stream");
    append_bytes(&runtime, &stream, b"cd").await;
    cold_store
        .write_chunk("benchcmp/external-index/external/tail.bin", b"ef")
        .await
        .expect("write tail external payload");
    runtime
        .append_external(AppendExternalRequest {
            stream_id: stream.clone(),
            content_type: DEFAULT_CONTENT_TYPE.to_owned(),
            payload: ExternalPayloadRef {
                s3_path: "benchcmp/external-index/external/tail.bin".to_owned(),
                payload_len: 2,
                object_size: 2,
            },
            record_ends: Vec::new(),
            close_after: false,
            stream_seq: None,
            producer: None,
            now_ms: 0,
            record_match: None,
        })
        .await
        .expect("append external payload");

    let read = runtime
        .read_stream(read_req(stream.clone(), 0, 6))
        .await
        .expect("read mixed external and hot payload");
    assert_eq!(read.payload, b"abcdef");
    assert_eq!(read.next_offset, 6);

    let snapshot = runtime
        .snapshot_group(runtime.locate(&stream).raft_group_id)
        .await
        .expect("snapshot group");
    let entry = snapshot
        .stream_snapshot
        .streams
        .iter()
        .find(|entry| entry.metadata.stream_id == stream)
        .expect("snapshot entry");
    assert_eq!(entry.cold_frontier_offset, 6);
    assert!(entry.cold_chunks.is_empty());
    assert!(entry.external_segments.is_empty());
}

#[tokio::test]
async fn bootstrap_reads_retained_updates_from_cold_chunk_after_snapshot() {
    let placement = placement();
    let stream = BucketStreamId::new("benchcmp", "cold-bootstrap");
    let cold_store = Arc::new(memory_cold_store());
    cold_store
        .write_chunk("benchcmp/cold-bootstrap/chunks/000000.bin", b"abcde")
        .await
        .expect("write cold object");
    let mut engine = InMemoryGroupEngine::with_cold_store(cold_store);

    engine
        .create_stream(
            CreateStreamRequest::new(stream.clone(), DEFAULT_CONTENT_TYPE),
            placement,
            ColdWriteAdmission::default(),
        )
        .await
        .expect("create stream");
    engine
        .append(
            AppendRequest::from_bytes(stream.clone(), b"abc".to_vec()),
            placement,
            ColdWriteAdmission::default(),
        )
        .await
        .expect("append first message");
    engine
        .append(
            AppendRequest::from_bytes(stream.clone(), b"de".to_vec()),
            placement,
            ColdWriteAdmission::default(),
        )
        .await
        .expect("append second message");
    engine
        .flush_cold(
            FlushColdRequest {
                stream_id: stream.clone(),
                chunk: ColdChunkRef {
                    start_offset: 0,
                    end_offset: 5,
                    s3_path: "benchcmp/cold-bootstrap/chunks/000000.bin".to_owned(),
                    object_size: 5,
                    ..Default::default()
                },
            },
            placement,
        )
        .await
        .expect("flush all hot bytes");
    engine
        .publish_snapshot(
            PublishSnapshotRequest {
                stream_id: stream.clone(),
                snapshot_offset: 3,
                content_type: DEFAULT_CONTENT_TYPE.to_owned(),
                payload: Bytes::from_static(b"abc-state"),
                expected_digest: None,
                now_ms: 0,
            },
            placement,
        )
        .await
        .expect("publish snapshot");
    engine
        .advance_retention(
            AdvanceRetentionRequest {
                stream_id: stream.clone(),
                retained_offset: 3,
                now_ms: 0,
            },
            placement,
        )
        .await
        .expect("advance retention");

    let read = engine
        .read_stream(read_req(stream.clone(), 3, 2), placement)
        .await
        .expect("read retained update from cold chunk");
    assert_eq!(read.payload, b"de");

    let bootstrap = engine
        .bootstrap_stream(
            BootstrapStreamRequest {
                stream_id: stream,
                now_ms: 0,
            },
            placement,
        )
        .await
        .expect("bootstrap");
    assert_eq!(bootstrap.snapshot_offset, Some(3));
    assert_eq!(bootstrap.snapshot_payload, b"abc-state");
    assert_eq!(bootstrap.next_offset, 5);
    assert_eq!(bootstrap.updates.len(), 1);
    assert_eq!(bootstrap.updates[0].start_offset, 3);
    assert_eq!(bootstrap.updates[0].next_offset, 5);
    assert_eq!(bootstrap.updates[0].payload, b"de");
}

#[tokio::test]
async fn cold_store_reads_only_requested_range() {
    let cold_store = memory_cold_store();
    cold_store
        .write_chunk("benchcmp/cold-range/chunks/000000.bin", b"abcdefgh")
        .await
        .expect("write cold object");
    let bytes = cold_store
        .read_chunk_range(
            &ColdChunkRef {
                start_offset: 10,
                end_offset: 18,
                s3_path: "benchcmp/cold-range/chunks/000000.bin".to_owned(),
                object_size: 8,
                ..Default::default()
            },
            12,
            3,
        )
        .await
        .expect("read range");
    assert_eq!(bytes, b"cde");
}

#[tokio::test]
async fn cold_store_prefetches_sequential_stream_blocks() {
    let cold_store = memory_cold_store().with_read_cache(ColdReadCacheParams {
        max_bytes: 32,
        block_bytes: 4,
        max_readahead_blocks: 2,
    });
    let path = "benchcmp/cold-cache/chunks/000000.bin";
    cold_store
        .write_chunk(path, b"abcdefghijklmnop")
        .await
        .expect("write cold object");
    let stream = BucketStreamId::new("benchcmp", "cold-cache");
    let object = ObjectPayloadRef {
        start_offset: 0,
        end_offset: 16,
        s3_path: path.to_owned(),
        object_size: 16,
        ..Default::default()
    };

    let first = cold_store
        .read_object_range_for_stream(&stream, &object, 0, 4)
        .await
        .expect("read first block");
    assert_eq!(first, b"abcd");
    let second = cold_store
        .read_object_range_for_stream(&stream, &object, 4, 4)
        .await
        .expect("read second block");
    assert_eq!(second, b"efgh");

    for _ in 0..20 {
        if cold_store.cached_block_count() >= 3 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    assert!(cold_store.cached_block_count() >= 3);

    cold_store
        .delete_chunk(path)
        .await
        .expect("delete cold object");
    assert_eq!(cold_store.cached_block_count(), 0);
}

#[tokio::test]
async fn ttl_read_access_is_committed_and_expiry_removes_stream() {
    let placement = placement();
    let stream = BucketStreamId::new("benchcmp", "runtime-ttl");
    let mut engine = InMemoryGroupEngine::default();

    let mut create = CreateStreamRequest::new(stream.clone(), DEFAULT_CONTENT_TYPE);
    create.initial_payload = Bytes::from_static(b"abc");
    create.stream_ttl_seconds = Some(1);
    create.now_ms = 1_000;
    engine
        .create_stream(create, placement, ColdWriteAdmission::default())
        .await
        .expect("create ttl stream");

    let read = engine
        .read_stream(
            ReadStreamRequest {
                stream_id: stream.clone(),
                offset: 0,
                max_len: 16,
                now_ms: 1_500,
                record: None,
                max_records: None,
                leader_only: false,
            },
            placement,
        )
        .await
        .expect("read renews ttl");
    assert_eq!(read.payload, b"abc");
    assert_eq!(
        engine
            .snapshot(placement)
            .await
            .expect("snapshot")
            .group_commit_index,
        2
    );

    engine
        .head_stream(
            HeadStreamRequest {
                stream_id: stream.clone(),
                now_ms: 2_499,
            },
            placement,
        )
        .await
        .expect("head does not renew but stream is still live");
    assert_eq!(
        engine
            .snapshot(placement)
            .await
            .expect("snapshot")
            .group_commit_index,
        2
    );

    let err = engine
        .read_stream(
            ReadStreamRequest {
                stream_id: stream.clone(),
                offset: 0,
                max_len: 16,
                now_ms: 2_500,
                record: None,
                max_records: None,
                leader_only: false,
            },
            placement,
        )
        .await
        .expect_err("expired stream read is not found");
    assert_eq!(err.code(), Some(StreamErrorCode::StreamNotFound));
    assert_eq!(
        engine
            .snapshot(placement)
            .await
            .expect("snapshot")
            .group_commit_index,
        3
    );

    let mut recreate = CreateStreamRequest::new(stream, "text/plain");
    recreate.now_ms = 2_501;
    let recreated = engine
        .create_stream(recreate, placement, ColdWriteAdmission::default())
        .await
        .expect("recreate expired stream");
    assert!(!recreated.already_exists);
}

#[test]
fn committed_write_batch_preserves_logical_command_responses() {
    let placement = placement();
    let stream = BucketStreamId::new("benchcmp", "apply-command-batch");
    let mut engine = InMemoryGroupEngine::default();

    let response = engine
        .apply_committed_write(
            GroupWriteCommand::Batch {
                commands: vec![
                    StreamCommand::from(CreateStreamRequest::new(
                        stream.clone(),
                        DEFAULT_CONTENT_TYPE,
                    )),
                    StreamCommand::from(AppendBatchRequest::new(stream.clone(), vec![
                        Bytes::from_static(b"ab"),
                        Bytes::from_static(b"cd"),
                    ])),
                ],
            },
            placement,
        )
        .expect("apply command batch");

    let GroupWriteResponse::Batch(items) = response else {
        panic!("unexpected batch response: {response:?}");
    };
    assert_eq!(items.len(), 2);
    assert!(matches!(
        &items[0],
        Ok(GroupWriteResponse::CreateStream(CreateStreamResponse {
            group_commit_index: 1,
            ..
        }))
    ));
    match &items[1] {
        Ok(GroupWriteResponse::AppendBatch(response)) => {
            assert_eq!(response.items.len(), 2);
            assert_eq!(
                response.items[0].as_ref().expect("first item").start_offset,
                0
            );
            assert_eq!(
                response.items[1]
                    .as_ref()
                    .expect("second item")
                    .start_offset,
                2
            );
            assert_eq!(
                response.items[1]
                    .as_ref()
                    .expect("second item")
                    .group_commit_index,
                3
            );
        }
        other => panic!("unexpected append batch response: {other:?}"),
    }

    let read = engine
        .state_machine
        .read(&stream, 0, 16)
        .expect("read applied command batch");
    assert_eq!(read.payload, b"abcd");
}

/// Poll `current` until it equals `expected` (up to ~1s), then panic with
/// `what` in the message if it never converges.
async fn wait_for_eq<T, F>(what: &str, expected: T, current: F)
where
    T: PartialEq + std::fmt::Display + Copy,
    F: Fn() -> T,
{
    for _ in 0..100 {
        if current() == expected {
            return;
        }
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
    panic!("expected {what} {expected}, got {}", current());
}

async fn wait_for_live_waiters(runtime: &ShardRuntime, expected: u64) {
    wait_for_eq("live waiters", expected, || {
        runtime.metrics().snapshot().live_read_waiters
    })
    .await;
}

async fn wait_for_mailbox_depth(runtime: &ShardRuntime, core_index: usize, expected: usize) {
    wait_for_eq(
        &format!("core {core_index} mailbox depth"),
        expected,
        || runtime.mailbox_snapshot().depths[core_index],
    )
    .await;
}

async fn wait_for_mailbox_full_events(runtime: &ShardRuntime, expected: u64) {
    wait_for_eq("mailbox full events", expected, || {
        runtime.metrics().snapshot().mailbox_full_events
    })
    .await;
}

async fn wait_for_group_mailbox_full_events(runtime: &ShardRuntime, expected: u64) {
    wait_for_eq("group mailbox full events", expected, || {
        runtime.metrics().snapshot().group_mailbox_full_events
    })
    .await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn repeated_appends_to_one_stream_are_ordered() {
    let runtime = runtime(4, 32);
    let stream = BucketStreamId::new("benchcmp", "one-stream");
    create_stream(&runtime, &stream).await;
    for index in 0..100 {
        let response = runtime
            .append(AppendRequest::new(stream.clone(), 7))
            .await
            .expect("append");
        assert_eq!(response.start_offset, index * 7);
        assert_eq!(response.next_offset, (index + 1) * 7);
        assert_eq!(response.stream_append_count, index + 1);
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn independent_streams_reach_all_cores_and_many_groups() {
    let runtime = runtime(4, 64);
    const STREAM_COUNT: usize = 4096;
    const MAX_IN_FLIGHT: usize = 64;

    for chunk_start in (0..STREAM_COUNT).step_by(MAX_IN_FLIGHT) {
        let chunk_end = (chunk_start + MAX_IN_FLIGHT).min(STREAM_COUNT);
        let mut tasks = Vec::with_capacity(chunk_end - chunk_start);
        for index in chunk_start..chunk_end {
            let runtime = runtime.clone();
            tasks.push(tokio::spawn(async move {
                let stream = BucketStreamId::new("benchcmp", format!("stream-{index}"));
                create_stream(&runtime, &stream).await;
                runtime
                    .append(AppendRequest::new(stream, 1))
                    .await
                    .expect("append")
            }));
        }

        for task in tasks {
            let response = task.await.expect("task");
            assert_eq!(response.start_offset, 0);
            assert_eq!(response.next_offset, 1);
        }
    }

    let snapshot = runtime.metrics().snapshot();
    assert_eq!(snapshot.accepted_appends, STREAM_COUNT as u64);
    assert!(snapshot.per_core_appends.iter().all(|value| *value > 0));
    let active_groups = snapshot
        .per_group_appends
        .iter()
        .filter(|value| **value > 0)
        .count();
    assert!(active_groups > 48, "active_groups={active_groups}");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn empty_append_is_rejected_before_routing() {
    let runtime = runtime(2, 8);
    let err = runtime
        .append(AppendRequest::new(BucketStreamId::new("b", "s"), 0))
        .await
        .expect_err("empty append rejected");
    assert_eq!(err, RuntimeError::EmptyAppend);
    assert_eq!(runtime.metrics().snapshot().accepted_appends, 0);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn append_batch_routes_once_and_applies_each_payload_on_owner_core() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "batch-runtime");
    let owner_core = usize::from(runtime.locate(&stream).core_id.0);
    let owner_group =
        usize::try_from(runtime.locate(&stream).raft_group_id.0).expect("u32 fits usize");

    create_stream(&runtime, &stream).await;
    let response = runtime
        .append_batch(AppendBatchRequest::new(stream.clone(), vec![
            b"ab".to_vec(),
            b"c".to_vec(),
            b"def".to_vec(),
        ]))
        .await
        .expect("append batch");
    assert_eq!(response.items.len(), 3);
    assert_eq!(response.items[0].as_ref().expect("first").start_offset, 0);
    assert_eq!(response.items[1].as_ref().expect("second").start_offset, 2);
    assert_eq!(response.items[2].as_ref().expect("third").start_offset, 3);

    let read = runtime
        .read_stream(read_req(stream.clone(), 0, 16))
        .await
        .expect("read");
    assert_eq!(read.payload, b"abcdef");

    let snapshot = runtime.metrics().snapshot();
    assert_eq!(snapshot.accepted_appends, 3);
    assert_eq!(snapshot.applied_mutations, 4);
    assert_eq!(snapshot.routed_requests, 3);
    assert_eq!(snapshot.per_core_appends[owner_core], 3);
    assert_eq!(snapshot.per_group_appends[owner_group], 3);
    assert_eq!(snapshot.per_core_applied_mutations[owner_core], 4);
    assert_eq!(snapshot.per_group_applied_mutations[owner_group], 4);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn append_batch_reports_item_errors_without_stopping_later_payloads() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "batch-partial");
    create_stream(&runtime, &stream).await;

    let response = runtime
        .append_batch(AppendBatchRequest::new(stream.clone(), vec![
            b"a".to_vec(),
            Vec::new(),
            b"b".to_vec(),
        ]))
        .await
        .expect("append batch");
    assert!(response.items[0].is_ok());
    assert!(response.items[1].is_err());
    assert!(response.items[2].is_ok());
    assert_eq!(response.items[0].as_ref().expect("first").start_offset, 0);
    assert_eq!(response.items[2].as_ref().expect("third").start_offset, 1);

    let read = runtime
        .read_stream(read_req(stream, 0, 16))
        .await
        .expect("read");
    assert_eq!(read.payload, b"ab");

    let snapshot = runtime.metrics().snapshot();
    assert_eq!(snapshot.accepted_appends, 2);
    assert_eq!(snapshot.applied_mutations, 3);
    assert_eq!(snapshot.routed_requests, 3);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn producer_duplicate_append_returns_prior_offsets_without_mutating_metrics() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "producer-runtime");
    create_stream(&runtime, &stream).await;

    let mut first = AppendRequest::from_bytes(stream.clone(), b"a".to_vec());
    first.producer = Some(producer("writer-1", 0, 0));
    let first = runtime.append(first).await.expect("first append");
    assert_eq!(first.start_offset, 0);
    assert_eq!(first.next_offset, 1);
    assert_eq!(first.stream_append_count, 1);
    assert!(!first.deduplicated);

    let mut duplicate = AppendRequest::from_bytes(stream.clone(), b"ignored".to_vec());
    duplicate.producer = Some(producer("writer-1", 0, 0));
    let duplicate = runtime.append(duplicate).await.expect("duplicate append");
    assert_eq!(duplicate.start_offset, 0);
    assert_eq!(duplicate.next_offset, 1);
    assert_eq!(duplicate.stream_append_count, 1);
    assert!(duplicate.deduplicated);

    let mut next = AppendRequest::from_bytes(stream.clone(), b"b".to_vec());
    next.producer = Some(producer("writer-1", 0, 1));
    let next = runtime.append(next).await.expect("next append");
    assert_eq!(next.start_offset, 1);
    assert_eq!(next.next_offset, 2);
    assert_eq!(next.stream_append_count, 2);
    assert!(!next.deduplicated);

    let read = runtime
        .read_stream(read_req(stream, 0, 16))
        .await
        .expect("read");
    assert_eq!(read.payload, b"ab");

    let metrics = runtime.metrics().snapshot();
    assert_eq!(metrics.accepted_appends, 2);
    assert_eq!(metrics.applied_mutations, 3);
    assert_eq!(metrics.routed_requests, 5);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn producer_duplicate_append_batch_returns_prior_offsets_without_mutating_metrics() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "producer-batch-runtime");
    create_stream(&runtime, &stream).await;

    let mut first = AppendBatchRequest::new(stream.clone(), vec![b"ab".to_vec(), b"c".to_vec()]);
    first.producer = Some(producer("writer-1", 0, 0));
    let first = runtime.append_batch(first).await.expect("first batch");
    assert_eq!(first.items.len(), 2);
    let first_item = first.items[0].as_ref().expect("first item");
    let second_item = first.items[1].as_ref().expect("second item");
    assert_eq!(first_item.start_offset, 0);
    assert_eq!(first_item.next_offset, 2);
    assert_eq!(first_item.stream_append_count, 1);
    assert!(!first_item.deduplicated);
    assert_eq!(second_item.start_offset, 2);
    assert_eq!(second_item.next_offset, 3);
    assert_eq!(second_item.stream_append_count, 2);
    assert!(!second_item.deduplicated);

    let mut duplicate =
        AppendBatchRequest::new(stream.clone(), vec![b"ignored".to_vec(), b"body".to_vec()]);
    duplicate.producer = Some(producer("writer-1", 0, 0));
    let duplicate = runtime
        .append_batch(duplicate)
        .await
        .expect("duplicate batch");
    assert_eq!(duplicate.items.len(), 2);
    assert!(
        duplicate
            .items
            .iter()
            .all(|item| { item.as_ref().expect("deduplicated item").deduplicated })
    );
    assert_eq!(
        duplicate.items[0]
            .as_ref()
            .expect("first duplicate")
            .start_offset,
        0
    );
    assert_eq!(
        duplicate.items[1]
            .as_ref()
            .expect("second duplicate")
            .next_offset,
        3
    );

    let mut next = AppendBatchRequest::new(stream.clone(), vec![b"d".to_vec()]);
    next.producer = Some(producer("writer-1", 0, 1));
    let next = runtime.append_batch(next).await.expect("next batch");
    let next_item = next.items[0].as_ref().expect("next item");
    assert_eq!(next_item.start_offset, 3);
    assert_eq!(next_item.next_offset, 4);
    assert_eq!(next_item.stream_append_count, 3);
    assert!(!next_item.deduplicated);

    let read = runtime
        .read_stream(read_req(stream, 0, 16))
        .await
        .expect("read");
    assert_eq!(read.payload, b"abcd");

    let metrics = runtime.metrics().snapshot();
    assert_eq!(metrics.accepted_appends, 3);
    assert_eq!(metrics.applied_mutations, 4);
    assert_eq!(metrics.routed_requests, 5);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn snapshot_group_routes_to_owner_core_and_captures_only_group_state() {
    let runtime = runtime(2, 8);
    let first_stream = BucketStreamId::new("benchcmp", "snapshot-first");
    let first_placement = runtime.locate(&first_stream);
    let second_stream = (0..512)
        .map(|index| BucketStreamId::new("benchcmp", format!("snapshot-other-{index}")))
        .find(|stream| runtime.locate(stream).core_id != first_placement.core_id)
        .expect("stream on another core");

    create_stream(&runtime, &first_stream).await;
    append_bytes(&runtime, &first_stream, b"first").await;
    create_stream(&runtime, &second_stream).await;
    append_bytes(&runtime, &second_stream, b"second").await;

    let snapshot = runtime
        .snapshot_group(first_placement.raft_group_id)
        .await
        .expect("snapshot group");
    assert_eq!(snapshot.placement, first_placement);
    assert_eq!(snapshot.group_commit_index, 2);
    assert_eq!(snapshot.stream_snapshot.buckets, vec!["benchcmp"]);
    assert_eq!(
        snapshot
            .stream_snapshot
            .streams
            .iter()
            .map(|entry| entry.metadata.stream_id.clone())
            .collect::<Vec<_>>(),
        vec![first_stream.clone()]
    );

    let restored =
        StreamStateMachine::restore(snapshot.stream_snapshot).expect("restore group snapshot");
    let read = restored
        .read(&first_stream, 0, 16)
        .expect("read restored snapshot");
    assert_eq!(read.payload, b"first");
    assert_eq!(read.next_offset, 5);
    assert!(restored.read(&second_stream, 0, 16).is_err());

    let metrics = runtime.metrics().snapshot();
    assert_eq!(metrics.routed_requests, 5);
    assert_eq!(
        metrics.per_core_routed_requests[usize::from(first_placement.core_id.0)],
        3
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn snapshot_group_rejects_out_of_range_group_before_routing() {
    let runtime = runtime(2, 8);
    let err = runtime
        .snapshot_group(RaftGroupId(8))
        .await
        .expect_err("invalid group");
    assert_eq!(err, RuntimeError::InvalidRaftGroup {
        raft_group_id: RaftGroupId(8),
        raft_group_count: 8,
    });
    assert_eq!(runtime.metrics().snapshot().routed_requests, 0);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn install_group_snapshot_restores_group_state_and_append_counts() {
    let source = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "install-snapshot");
    let placement = source.locate(&stream);
    let attrs = stream_attrs("Snapshot session", "snapshot-install");
    create_stream(&source, &stream).await;
    source
        .update_stream_attrs(UpdateStreamAttrsRequest {
            stream_id: stream.clone(),
            attrs: Some(attrs.clone()),
            now_ms: 0,
        })
        .await
        .expect("update attrs before snapshot");
    append_bytes(&source, &stream, b"ab").await;
    append_bytes(&source, &stream, b"cd").await;

    let snapshot = source
        .snapshot_group(placement.raft_group_id)
        .await
        .expect("snapshot group");
    assert_eq!(snapshot.group_commit_index, 4);
    let snapshot_commit_index = snapshot.group_commit_index;
    assert_eq!(snapshot.stream_append_counts, vec![StreamAppendCount {
        stream_id: stream.clone(),
        append_count: 2,
    }]);

    let target = runtime(2, 8);
    target
        .install_group_snapshot(snapshot)
        .await
        .expect("install snapshot");

    let read = target
        .read_stream(read_req(stream.clone(), 0, 16))
        .await
        .expect("read restored stream");
    assert_eq!(read.placement, placement);
    assert_eq!(read.payload, b"abcd");
    assert_eq!(read.next_offset, 4);
    let restored_attrs = target
        .get_stream_attrs(GetStreamAttrsRequest {
            stream_id: stream.clone(),
            now_ms: 0,
        })
        .await
        .expect("get restored attrs");
    assert_eq!(restored_attrs.attrs, Some(attrs));

    let appended = target
        .append(AppendRequest::from_bytes(stream, b"ef".to_vec()))
        .await
        .expect("append after restore");
    assert_eq!(appended.start_offset, 4);
    assert_eq!(appended.next_offset, 6);
    assert_eq!(appended.stream_append_count, 3);
    assert_eq!(appended.group_commit_index, snapshot_commit_index + 1);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn snapshot_after_stream_delete_installs_without_dangling_append_count() {
    // Regression: a deleted stream must not leave a stale entry in the runtime
    // append-count map. Such an entry would make the snapshot fail every
    // follower's install with "append count references missing snapshot stream",
    // so a lagging node could never catch up and leadership transfer (which
    // catches the target up via a snapshot) could never complete.
    let source = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "churn-delete");
    let placement = source.locate(&stream);
    create_stream(&source, &stream).await;
    append_bytes(&source, &stream, b"ab").await;
    source
        .delete_stream(DeleteStreamRequest {
            stream_id: stream.clone(),
        })
        .await
        .expect("delete");

    let snapshot = source
        .snapshot_group(placement.raft_group_id)
        .await
        .expect("snapshot group");
    assert!(
        snapshot
            .stream_append_counts
            .iter()
            .all(|count| count.stream_id != stream),
        "snapshot must not carry an append count for the deleted stream: {:?}",
        snapshot.stream_append_counts
    );

    // Previously this failed the restore consistency check.
    let target = runtime(2, 8);
    target
        .install_group_snapshot(snapshot)
        .await
        .expect("install snapshot after delete");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn install_group_snapshot_rejects_mismatched_placement_before_routing() {
    let runtime = runtime(2, 8);
    let snapshot = GroupSnapshot {
        placement: ShardPlacement {
            core_id: CoreId(1),
            shard_id: ShardId(0),
            raft_group_id: RaftGroupId(0),
        },
        group_commit_index: 0,
        stream_snapshot: StreamSnapshot {
            buckets: Vec::new(),
            erased_buckets: Vec::new(),
            streams: Vec::new(),
            pending_cold_gc: Vec::new(),
            next_cold_gc_seq: 0,
            shared_cold_object_owners: Vec::new(),
            bucket_usage: Vec::new(),
            bucket_quotas: Vec::new(),
        },
        stream_append_counts: Vec::new(),
    };

    let err = runtime
        .install_group_snapshot(snapshot)
        .await
        .expect_err("mismatched placement rejected");
    assert_eq!(err, RuntimeError::SnapshotPlacementMismatch {
        expected: ShardPlacement {
            core_id: CoreId(0),
            shard_id: ShardId(0),
            raft_group_id: RaftGroupId(0),
        },
        actual: ShardPlacement {
            core_id: CoreId(1),
            shard_id: ShardId(0),
            raft_group_id: RaftGroupId(0),
        },
    });
    assert_eq!(runtime.metrics().snapshot().routed_requests, 0);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn mailbox_snapshot_reports_per_core_depths_and_capacities() {
    let runtime = ShardRuntime::spawn(test_config(3, 9, 7)).expect("spawn runtime");

    let snapshot = runtime.mailbox_snapshot();
    assert_eq!(snapshot.depths, vec![0, 0, 0]);
    assert_eq!(snapshot.capacities, vec![7, 7, 7]);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn runtime_metrics_track_owner_core_routing_and_mailbox_wait() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "routing-metrics");
    let owner_core = usize::from(runtime.locate(&stream).core_id.0);

    create_stream(&runtime, &stream).await;
    append_bytes(&runtime, &stream, b"hello").await;
    runtime
        .read_stream(read_req(stream.clone(), 0, 16))
        .await
        .expect("read");

    let snapshot = runtime.metrics().snapshot();
    assert_eq!(snapshot.accepted_appends, 1);
    assert_eq!(snapshot.applied_mutations, 2);
    assert_eq!(snapshot.routed_requests, 3);
    assert_eq!(snapshot.per_core_routed_requests.len(), 2);
    assert_eq!(snapshot.per_core_routed_requests[owner_core], 3);
    assert_eq!(snapshot.per_core_applied_mutations[owner_core], 2);
    assert_eq!(
        snapshot.per_group_applied_mutations
            [usize::try_from(runtime.locate(&stream).raft_group_id.0).expect("u32 fits usize")],
        2
    );
    assert_eq!(
        snapshot.mutation_apply_ns,
        snapshot.per_core_mutation_apply_ns.iter().sum::<u64>()
    );
    assert_eq!(
        snapshot.mutation_apply_ns,
        snapshot.per_group_mutation_apply_ns.iter().sum::<u64>()
    );
    assert_eq!(
        snapshot.group_lock_wait_ns,
        snapshot.per_core_group_lock_wait_ns.iter().sum::<u64>()
    );
    assert_eq!(
        snapshot.group_lock_wait_ns,
        snapshot.per_group_group_lock_wait_ns.iter().sum::<u64>()
    );
    assert_eq!(
        snapshot.group_engine_exec_ns,
        snapshot.per_core_group_engine_exec_ns.iter().sum::<u64>()
    );
    assert_eq!(
        snapshot.group_engine_exec_ns,
        snapshot.per_group_group_engine_exec_ns.iter().sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_write_many_batches,
        snapshot
            .per_core_raft_write_many_batches
            .iter()
            .sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_write_many_batches,
        snapshot
            .per_group_raft_write_many_batches
            .iter()
            .sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_write_many_commands,
        snapshot
            .per_core_raft_write_many_commands
            .iter()
            .sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_write_many_commands,
        snapshot
            .per_group_raft_write_many_commands
            .iter()
            .sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_write_many_logical_commands,
        snapshot
            .per_core_raft_write_many_logical_commands
            .iter()
            .sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_write_many_logical_commands,
        snapshot
            .per_group_raft_write_many_logical_commands
            .iter()
            .sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_write_many_responses,
        snapshot
            .per_core_raft_write_many_responses
            .iter()
            .sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_write_many_responses,
        snapshot
            .per_group_raft_write_many_responses
            .iter()
            .sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_write_many_submit_ns,
        snapshot
            .per_core_raft_write_many_submit_ns
            .iter()
            .sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_write_many_submit_ns,
        snapshot
            .per_group_raft_write_many_submit_ns
            .iter()
            .sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_write_many_response_ns,
        snapshot
            .per_core_raft_write_many_response_ns
            .iter()
            .sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_write_many_response_ns,
        snapshot
            .per_group_raft_write_many_response_ns
            .iter()
            .sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_apply_entries,
        snapshot.per_core_raft_apply_entries.iter().sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_apply_entries,
        snapshot.per_group_raft_apply_entries.iter().sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_apply_ns,
        snapshot.per_core_raft_apply_ns.iter().sum::<u64>()
    );
    assert_eq!(
        snapshot.raft_apply_ns,
        snapshot.per_group_raft_apply_ns.iter().sum::<u64>()
    );
    assert_eq!(
        snapshot.mailbox_send_wait_ns,
        snapshot.per_core_mailbox_send_wait_ns.iter().sum::<u64>()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn append_before_stream_setup_uses_stream_state_machine_error() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "missing-stream");
    let placement = runtime.locate(&stream);
    let err = runtime
        .append(AppendRequest::new(stream, 1))
        .await
        .expect_err("missing stream rejected");

    match err {
        RuntimeError::GroupEngine {
            core_id,
            raft_group_id,
            error,
            ..
        } => {
            assert_eq!(core_id, placement.core_id);
            assert_eq!(raft_group_id, placement.raft_group_id);
            assert_eq!(error.code(), Some(StreamErrorCode::BucketNotFound));
            let message = error.message();
            assert!(message.contains("BucketNotFound"), "message={message}");
        }
        other => panic!("expected group engine error, got {other:?}"),
    }
    assert_eq!(runtime.metrics().snapshot().accepted_appends, 0);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn group_engine_errors_use_operation_wording_for_non_append_paths() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "missing-read-stream");
    let err = runtime
        .read_stream(read_req(stream, 0, 16))
        .await
        .expect_err("missing stream read rejected");
    let message = err.to_string();
    assert!(message.contains("operation failed"), "message={message}");
    assert!(!message.contains("append failed"), "message={message}");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn create_stream_is_routed_and_idempotent_for_matching_metadata() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "create-stream");
    let placement = runtime.locate(&stream);

    let created = create_stream(&runtime, &stream).await;
    assert_eq!(created.placement, placement);
    assert_eq!(created.next_offset, 0);
    assert!(!created.closed);
    assert!(!created.already_exists);

    let existing = create_stream(&runtime, &stream).await;
    assert_eq!(existing.placement, placement);
    assert_eq!(existing.next_offset, 0);
    assert!(!existing.closed);
    assert!(existing.already_exists);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn runtime_updates_and_reads_stream_attrs_on_owner_group() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "stream-attrs");
    create_stream(&runtime, &stream).await;
    let attrs = stream_attrs("Support session", "customer-support");

    let updated = runtime
        .update_stream_attrs(UpdateStreamAttrsRequest {
            stream_id: stream.clone(),
            attrs: Some(attrs.clone()),
            now_ms: 0,
        })
        .await
        .expect("update stream attrs");
    assert!(updated.changed);

    let read = runtime
        .get_stream_attrs(GetStreamAttrsRequest {
            stream_id: stream,
            now_ms: 0,
        })
        .await
        .expect("get stream attrs");
    assert_eq!(read.attrs, Some(attrs));
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn head_stream_reflects_append_and_closed_state_on_owner_group() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "head-stream");
    let placement = runtime.locate(&stream);
    runtime
        .create_stream(CreateStreamRequest::new(stream.clone(), "text/plain"))
        .await
        .expect("create stream");

    let mut append = AppendRequest::new(stream.clone(), 3);
    append.content_type = "text/plain".to_owned();
    append.close_after = true;
    let response = runtime.append(append).await.expect("append");
    assert_eq!(response.start_offset, 0);
    assert_eq!(response.next_offset, 3);

    let head = runtime
        .head_stream(HeadStreamRequest {
            stream_id: stream,
            now_ms: 0,
        })
        .await
        .expect("head stream");
    assert_eq!(head.placement, placement);
    assert_eq!(head.content_type, "text/plain");
    assert_eq!(head.tail_offset, 3);
    assert!(head.closed);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn read_stream_returns_payload_slice_from_owner_group() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "read-stream");
    let placement = runtime.locate(&stream);
    create_stream(&runtime, &stream).await;
    append_bytes(&runtime, &stream, b"abcdefg").await;

    let read = runtime
        .read_stream(read_req(stream.clone(), 2, 3))
        .await
        .expect("read stream");
    assert_eq!(read.placement, placement);
    assert_eq!(read.offset, 2);
    assert_eq!(read.next_offset, 5);
    assert_eq!(read.payload, b"cde");
    assert!(!read.up_to_date);
    assert!(!read.closed);

    let tail = runtime
        .read_stream(read_req(stream, 7, 16))
        .await
        .expect("tail read");
    assert_eq!(tail.next_offset, 7);
    assert!(tail.payload.is_empty());
    assert!(tail.up_to_date);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn flush_cold_publishes_chunk_metadata_on_owner_group() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "cold-runtime");
    let placement = runtime.locate(&stream);
    create_stream(&runtime, &stream).await;
    append_bytes(&runtime, &stream, b"abcdef").await;

    let flushed = runtime
        .flush_cold(FlushColdRequest {
            stream_id: stream.clone(),
            chunk: ColdChunkRef {
                start_offset: 0,
                end_offset: 4,
                s3_path: "s3://bucket/cold-runtime/000000".to_owned(),
                object_size: 4,
                ..Default::default()
            },
        })
        .await
        .expect("flush cold");
    assert_eq!(flushed.placement, placement);
    assert_eq!(flushed.hot_start_offset, 4);

    let hot = runtime
        .read_stream(read_req(stream.clone(), 4, 16))
        .await
        .expect("hot read");
    assert_eq!(hot.payload, b"ef");

    let err = runtime
        .read_stream(read_req(stream, 0, 16))
        .await
        .expect_err("cold read needs store");
    match err {
        RuntimeError::GroupEngine { error, .. } => {
            assert_eq!(error.code(), Some(StreamErrorCode::InvalidColdFlush));
            assert_eq!(error.next_offset(), Some(6));
        }
        other => panic!("expected cold read error, got {other:?}"),
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn flush_cold_once_uploads_outside_group_and_reads_back() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(RuntimeConfig::new(2, 8), cold_store);
    let stream = BucketStreamId::new("benchcmp", "cold-once");
    create_stream(&runtime, &stream).await;
    append_bytes(&runtime, &stream, b"abcdef").await;

    let flushed = runtime
        .flush_cold_once(PlanColdFlushRequest {
            stream_id: stream.clone(),
            min_hot_bytes: 4,
            max_flush_bytes: 4,
        })
        .await
        .expect("flush once")
        .expect("candidate flushed");
    assert_eq!(flushed.hot_start_offset, 4);
    let metrics = runtime.metrics().snapshot();
    assert_eq!(metrics.cold_flush_uploads, 1);
    assert_eq!(metrics.cold_flush_upload_bytes, 4);
    assert_eq!(metrics.cold_pack_uploads, 0);
    assert_eq!(metrics.cold_flush_publishes, 1);
    assert_eq!(metrics.cold_flush_publish_bytes, 4);
    assert_eq!(metrics.cold_orphan_cleanup_attempts, 0);

    let read = runtime
        .read_stream(read_req(stream, 0, 6))
        .await
        .expect("read cold and hot");
    assert_eq!(read.payload, b"abcdef");
    assert_eq!(read.next_offset, 6);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn flush_cold_group_batch_once_publishes_multiple_chunks() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(RuntimeConfig::new(2, 8), cold_store);
    let stream = BucketStreamId::new("benchcmp", "cold-batch");
    let placement = runtime.locate(&stream);
    create_stream(&runtime, &stream).await;
    append_bytes(&runtime, &stream, b"abcd").await;

    let flushed = runtime
        .flush_cold_group_batch_once(
            placement.raft_group_id,
            PlanGroupColdFlushRequest {
                min_hot_bytes: 1,
                max_flush_bytes: 1,
                max_batch_bytes: 4,
            },
            4,
        )
        .await
        .expect("flush batch");
    assert_eq!(flushed.len(), 4);
    assert!(
        flushed
            .iter()
            .all(|response| response.placement == placement)
    );
    assert_eq!(
        flushed
            .iter()
            .map(|response| response.hot_start_offset)
            .collect::<Vec<_>>(),
        vec![1, 2, 3, 4]
    );

    let metrics = runtime.metrics().snapshot();
    assert_eq!(metrics.cold_flush_uploads, 1);
    assert_eq!(metrics.cold_flush_upload_bytes, 4);
    assert_eq!(metrics.cold_pack_uploads, 1);
    assert_eq!(metrics.cold_pack_bytes, 4);
    assert_eq!(metrics.cold_pack_slices, 4);
    assert_eq!(metrics.cold_flush_publishes, 4);
    assert_eq!(metrics.cold_flush_publish_bytes, 4);
    assert_eq!(metrics.cold_hot_bytes, 0);

    let snapshot = runtime
        .snapshot_group(placement.raft_group_id)
        .await
        .expect("snapshot group");
    let entry = snapshot
        .stream_snapshot
        .streams
        .iter()
        .find(|entry| entry.metadata.stream_id == stream)
        .expect("stream snapshot");
    assert_eq!(entry.cold_frontier_offset, 4);
    assert_eq!(entry.cold_chunks.len(), 4);
    assert!(
        entry
            .cold_chunks
            .iter()
            .all(|chunk| chunk.shared_object && chunk.object_size == 4)
    );
    assert_eq!(
        entry
            .cold_chunks
            .iter()
            .map(|chunk| chunk.object_offset)
            .collect::<Vec<_>>(),
        vec![0, 1, 2, 3]
    );
    assert!(entry.payload.is_empty());

    let read = runtime
        .read_stream(read_req(stream, 0, 4))
        .await
        .expect("read cold chunks");
    assert_eq!(read.payload, b"abcd");
    assert_eq!(read.next_offset, 4);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn packed_cold_object_survives_until_last_stream_is_deleted() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(RuntimeConfig::new(2, 8), cold_store.clone());
    let group_id = RaftGroupId(3);
    let first = stream_on_group(&runtime, group_id, "pack-first");
    let second = stream_on_group(&runtime, group_id, "pack-second");
    assert_ne!(first, second);
    for (stream, payload) in [(&first, b"abcd".as_slice()), (&second, b"efgh".as_slice())] {
        create_stream(&runtime, stream).await;
        append_bytes(&runtime, stream, payload).await;
    }

    let flushed = runtime
        .flush_cold_group_batch_once(
            group_id,
            PlanGroupColdFlushRequest {
                min_hot_bytes: 4,
                max_flush_bytes: 4,
                max_batch_bytes: 8,
            },
            8,
        )
        .await
        .expect("flush shared pack");
    assert_eq!(flushed.len(), 2);
    assert_eq!(runtime.metrics().snapshot().cold_flush_uploads, 1);

    let snapshot = runtime
        .snapshot_group(group_id)
        .await
        .expect("snapshot packed group");
    let chunks = snapshot
        .stream_snapshot
        .streams
        .iter()
        .filter_map(|entry| entry.cold_chunks.first().cloned())
        .collect::<Vec<_>>();
    assert_eq!(chunks.len(), 2);
    assert_eq!(chunks[0].s3_path, chunks[1].s3_path);

    runtime
        .delete_stream(DeleteStreamRequest { stream_id: first })
        .await
        .expect("delete first packed stream");
    runtime
        .run_cold_gc_all_groups_once(256)
        .await
        .expect("gc first packed stream");
    assert_eq!(
        cold_store
            .read_chunk_range(&chunks[1], chunks[1].start_offset, 4)
            .await
            .expect("shared pack remains readable"),
        b"efgh"
    );

    runtime
        .delete_stream(DeleteStreamRequest { stream_id: second })
        .await
        .expect("delete second packed stream");
    runtime
        .run_cold_gc_all_groups_once(256)
        .await
        .expect("gc final packed stream");
    assert!(
        cold_store
            .read_chunk_range(&chunks[1], chunks[1].start_offset, 4)
            .await
            .is_err()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cold_packs_are_bucket_scoped_erasure_domains_within_one_raft_group() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(RuntimeConfig::new(2, 8), cold_store.clone());
    let group_id = RaftGroupId(3);
    let bucket_a = "erasure-a";
    let bucket_b = "erasure-b";
    let streams = [
        stream_in_bucket_on_group(&runtime, group_id, bucket_a, "a-first"),
        stream_in_bucket_on_group(&runtime, group_id, bucket_a, "a-second"),
        stream_in_bucket_on_group(&runtime, group_id, bucket_b, "b-first"),
        stream_in_bucket_on_group(&runtime, group_id, bucket_b, "b-second"),
    ];
    for (index, stream) in streams.iter().enumerate() {
        create_stream(&runtime, stream).await;
        append_bytes(
            &runtime,
            stream,
            &[b'a' + u8::try_from(index).expect("small index"); 4],
        )
        .await;
    }

    let flushed = runtime
        .flush_cold_group_batch_once(
            group_id,
            PlanGroupColdFlushRequest {
                min_hot_bytes: 4,
                max_flush_bytes: 4,
                max_batch_bytes: 16,
            },
            16,
        )
        .await
        .expect("flush bucket-scoped packs");
    assert_eq!(flushed.len(), 4);
    assert_eq!(runtime.metrics().snapshot().cold_flush_uploads, 2);

    let snapshot = runtime
        .snapshot_group(group_id)
        .await
        .expect("snapshot packed group");
    let chunk_for = |stream: &BucketStreamId| {
        snapshot
            .stream_snapshot
            .streams
            .iter()
            .find(|entry| entry.metadata.stream_id == *stream)
            .and_then(|entry| entry.cold_chunks.first())
            .cloned()
            .expect("cold chunk")
    };
    let a_chunks = streams[..2].iter().map(chunk_for).collect::<Vec<_>>();
    let b_chunks = streams[2..].iter().map(chunk_for).collect::<Vec<_>>();
    assert_eq!(a_chunks[0].s3_path, a_chunks[1].s3_path);
    assert_eq!(b_chunks[0].s3_path, b_chunks[1].s3_path);
    assert_ne!(a_chunks[0].s3_path, b_chunks[0].s3_path);
    assert!(a_chunks[0].s3_path.starts_with("erasure-a/_packs/"));
    assert!(b_chunks[0].s3_path.starts_with("erasure-b/_packs/"));

    let purge = runtime
        .purge_bucket_all_groups(bucket_a)
        .await
        .expect("purge first bucket");
    assert_eq!(purge.removed_streams, 2);
    assert_eq!(purge.pending_cold_gc_entries, 3);
    runtime
        .run_cold_gc_all_groups_once(256)
        .await
        .expect("reclaim first bucket");
    let proof = runtime
        .purge_bucket_all_groups(bucket_a)
        .await
        .expect("prove first bucket absence");
    assert_eq!(proof.pending_cold_gc_entries, 0);
    assert!(
        cold_store
            .read_chunk_range(&a_chunks[0], a_chunks[0].start_offset, 4)
            .await
            .is_err()
    );
    assert_eq!(
        cold_store
            .read_chunk_range(&b_chunks[0], b_chunks[0].start_offset, 4)
            .await
            .expect("other bucket pack remains readable"),
        b"cccc"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn legacy_cross_bucket_pack_is_rewritten_before_bucket_erasure_proof() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(RuntimeConfig::new(2, 8), cold_store.clone());
    let group_id = RaftGroupId(3);
    let stream_a = stream_in_bucket_on_group(&runtime, group_id, "legacy-erasure-a", "shared-a");
    let stream_b = stream_in_bucket_on_group(&runtime, group_id, "legacy-erasure-b", "shared-b");
    for stream in [&stream_a, &stream_b] {
        create_stream(&runtime, stream).await;
    }
    append_bytes(&runtime, &stream_a, b"aaaa").await;
    append_bytes(&runtime, &stream_b, b"bbbb").await;

    let legacy_path = "_packs/00000003/legacy-cross-bucket.bin";
    cold_store
        .write_chunk(legacy_path, b"aaaabbbb")
        .await
        .expect("write legacy pack");
    for (stream, object_offset, digest) in [
        (&stream_a, 0, blake3::hash(b"aaaa").to_hex().to_string()),
        (&stream_b, 4, blake3::hash(b"bbbb").to_hex().to_string()),
    ] {
        let chunk = ColdChunkRef {
            start_offset: 0,
            end_offset: 4,
            s3_path: legacy_path.to_owned(),
            object_size: 8,
            object_offset,
            shared_object: true,
            payload_digest: digest,
        };
        runtime
            .flush_cold(FlushColdRequest {
                stream_id: stream.clone(),
                chunk,
            })
            .await
            .expect("publish legacy shared slice");
    }
    let migration = runtime
        .migrate_legacy_shared_cold_once(1, 0)
        .await
        .expect("rewrite first legacy slice");
    assert_eq!(migration.observed_chunks, 2);
    assert_eq!(migration.migrated_chunks, 1);
    assert_eq!(migration.pending_chunks, 1);
    let intermediate = runtime
        .snapshot_group(group_id)
        .await
        .expect("snapshot partial legacy migration");
    assert_eq!(
        intermediate.stream_snapshot.shared_cold_object_owners.len(),
        1
    );
    assert_eq!(
        intermediate.stream_snapshot.shared_cold_object_owners[0].bucket_ids,
        vec!["legacy-erasure-a".to_owned(), "legacy-erasure-b".to_owned()]
    );

    let migration = runtime
        .migrate_legacy_shared_cold_once(2, 0)
        .await
        .expect("rewrite final legacy slice");
    assert_eq!(migration.observed_chunks, 1);
    assert_eq!(migration.migrated_chunks, 1);
    assert_eq!(migration.pending_chunks, 0);
    runtime
        .run_cold_gc_all_groups_once(256)
        .await
        .expect("delete legacy pack for every bucket owner");
    let legacy_a = ColdChunkRef {
        start_offset: 0,
        end_offset: 4,
        s3_path: legacy_path.to_owned(),
        object_size: 8,
        object_offset: 0,
        shared_object: true,
        ..Default::default()
    };
    assert!(
        cold_store.read_chunk_range(&legacy_a, 0, 4).await.is_err(),
        "the cross-bucket physical object must be absent"
    );

    let page_store = ColdStoreColdIndexPageStore::new(cold_store.clone());
    async fn rewritten(
        page_store: &ColdStoreColdIndexPageStore,
        stream: &BucketStreamId,
    ) -> ColdChunkRef {
        load_cold_chunks_from_pages(page_store, &[ColdIndexPageKey {
            stream_id: stream.clone(),
            generation: 0,
            page_id: 0,
        }])
        .await
        .expect("load rewritten index")
        .into_iter()
        .next()
        .expect("rewritten exclusive chunk")
    }
    let a_chunk = rewritten(&page_store, &stream_a).await;
    let b_chunk = rewritten(&page_store, &stream_b).await;
    assert!(!a_chunk.shared_object);
    assert!(!b_chunk.shared_object);
    assert!(a_chunk.s3_path.starts_with("legacy-erasure-a/"));
    assert!(b_chunk.s3_path.starts_with("legacy-erasure-b/"));

    let purge = runtime
        .purge_bucket_all_groups("legacy-erasure-a")
        .await
        .expect("purge migrated bucket");
    assert_eq!(purge.removed_streams, 1);
    runtime
        .run_cold_gc_all_groups_once(256)
        .await
        .expect("erase migrated bucket");
    assert!(cold_store.read_chunk_range(&a_chunk, 0, 4).await.is_err());
    assert_eq!(
        cold_store
            .read_chunk_range(&b_chunk, 0, 4)
            .await
            .expect("other bucket remains readable"),
        b"bbbb"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn all_stale_packed_candidates_reclaim_unpublished_object() {
    let cold_store = Arc::new(memory_cold_store());
    let events = Arc::new(Mutex::new(Vec::new()));
    let observed = events.clone();
    cold_store.set_observer(move |event| {
        observed.lock().expect("event mutex").push(event);
    });
    let runtime = spawn_with_cold_store(RuntimeConfig::new(2, 8), cold_store);
    let group_id = RaftGroupId(3);
    let streams = [
        stream_on_group(&runtime, group_id, "stale-pack-first"),
        stream_on_group(&runtime, group_id, "stale-pack-second"),
    ];
    for stream in &streams {
        create_stream(&runtime, stream).await;
        append_bytes(&runtime, stream, b"abcd").await;
    }
    let candidates = runtime
        .plan_next_cold_flush_batch(
            group_id,
            PlanGroupColdFlushRequest {
                min_hot_bytes: 4,
                max_flush_bytes: 4,
                max_batch_bytes: 8,
            },
            8,
        )
        .await
        .expect("plan shared pack");
    assert_eq!(candidates.len(), 2);

    for stream in &streams {
        runtime
            .delete_stream(DeleteStreamRequest {
                stream_id: stream.clone(),
            })
            .await
            .expect("delete planned stream");
        create_stream(&runtime, stream).await;
        append_bytes(&runtime, stream, b"new!").await;
    }
    assert!(
        runtime
            .flush_cold_candidates_batch(candidates)
            .await
            .expect("stale pack is classified")
            .is_empty()
    );

    let events = events.lock().expect("event mutex");
    let written = events.iter().find_map(|event| match event {
        ColdStoreEvent::WriteChunkComplete { path, .. } if path.starts_with("benchcmp/_packs/") => {
            Some(path)
        }
        _ => None,
    });
    let deleted = events.iter().find_map(|event| match event {
        ColdStoreEvent::DeleteChunkComplete { path } if path.starts_with("benchcmp/_packs/") => {
            Some(path)
        }
        _ => None,
    });
    assert!(written.is_some());
    assert_eq!(written, deleted);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cold_gc_worker_physically_reclaims_deleted_stream_chunks() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(RuntimeConfig::new(2, 8), cold_store.clone());
    let stream = BucketStreamId::new("benchcmp", "cold-gc");
    create_stream(&runtime, &stream).await;
    append_bytes(&runtime, &stream, b"abcd").await;
    let chunk = ColdChunkRef {
        start_offset: 0,
        end_offset: 4,
        s3_path: "benchcmp/cold-gc/chunks/000000.bin".to_owned(),
        object_size: 4,
        ..Default::default()
    };
    cold_store
        .write_chunk(&chunk.s3_path, b"abcd")
        .await
        .expect("write cold chunk");
    runtime
        .flush_cold(FlushColdRequest {
            stream_id: stream.clone(),
            chunk: chunk.clone(),
        })
        .await
        .expect("flush cold");

    assert!(
        cold_store
            .read_chunk_range(&chunk, chunk.start_offset, 4)
            .await
            .is_ok(),
        "chunk must exist before GC"
    );

    runtime
        .delete_stream(DeleteStreamRequest {
            stream_id: stream.clone(),
        })
        .await
        .expect("delete stream");

    // A GC tick reclaims exactly the one queued prefix and drains the queue.
    let reclaimed = runtime
        .run_cold_gc_all_groups_once(256)
        .await
        .expect("run cold gc");
    assert_eq!(reclaimed, 1);
    assert_eq!(runtime.metrics().snapshot().cold_gc_reclaimed, 1);

    // The physical object is gone, and a second tick finds nothing to do.
    assert!(
        cold_store
            .read_chunk_range(&chunk, chunk.start_offset, 4)
            .await
            .is_err(),
        "chunk must be physically reclaimed after GC"
    );
    assert_eq!(
        runtime
            .run_cold_gc_all_groups_once(256)
            .await
            .expect("idempotent gc tick"),
        0
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn purge_report_proves_cold_gc_queue_is_empty_only_after_reclamation() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(RuntimeConfig::new(2, 8), cold_store.clone());
    let stream = BucketStreamId::new("offboard-tenant", "cold-payload");
    create_stream(&runtime, &stream).await;
    append_bytes(&runtime, &stream, b"abcd").await;
    let chunk = ColdChunkRef {
        start_offset: 0,
        end_offset: 4,
        s3_path: "offboard-tenant/cold-payload/chunks/000000.bin".to_owned(),
        object_size: 4,
        ..Default::default()
    };
    cold_store
        .write_chunk(&chunk.s3_path, b"abcd")
        .await
        .expect("write cold chunk");
    runtime
        .flush_cold(FlushColdRequest {
            stream_id: stream,
            chunk: chunk.clone(),
        })
        .await
        .expect("flush cold");

    let purge = runtime
        .purge_bucket_all_groups("offboard-tenant")
        .await
        .expect("purge bucket");
    assert_eq!(purge.removed_streams, 1);
    assert_eq!(purge.pending_cold_gc_entries, 1);

    runtime
        .run_cold_gc_all_groups_once(256)
        .await
        .expect("reclaim cold payload");
    let proof = runtime
        .purge_bucket_all_groups("offboard-tenant")
        .await
        .expect("prove purge");
    assert_eq!(proof.removed_streams, 0);
    assert_eq!(proof.pending_cold_gc_entries, 0);
    assert!(
        cold_store
            .read_chunk_range(&chunk, chunk.start_offset, 4)
            .await
            .is_err()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cold_compaction_preserves_reads_and_reclaims_inputs_after_grace() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(RuntimeConfig::new(2, 8), cold_store.clone());
    let stream = BucketStreamId::new("benchcmp", "compact-runtime");
    create_stream(&runtime, &stream).await;
    append_bytes(&runtime, &stream, b"abcdefgh").await;
    for (start, end, payload) in [(0, 4, b"abcd".as_slice()), (4, 8, b"efgh".as_slice())] {
        let chunk = ColdChunkRef {
            start_offset: start,
            end_offset: end,
            s3_path: format!("benchcmp/compact-runtime/chunks/{start}.bin"),
            object_size: end - start,
            ..Default::default()
        };
        cold_store
            .write_chunk(&chunk.s3_path, payload)
            .await
            .expect("write input chunk");
        runtime
            .flush_cold(FlushColdRequest {
                stream_id: stream.clone(),
                chunk,
            })
            .await
            .expect("flush input chunk");
    }

    assert_eq!(
        runtime
            .compact_cold_once(8, 16, 1, 0)
            .await
            .expect("compact cold chunks"),
        1
    );
    let read = runtime
        .read_stream(read_req(stream.clone(), 0, 8))
        .await
        .expect("read compacted stream");
    assert_eq!(read.payload, b"abcdefgh");
    assert_eq!(
        runtime
            .run_cold_gc_all_groups_once(256)
            .await
            .expect("reclaim compacted inputs"),
        1
    );
    for start in [0, 4] {
        let old = ColdChunkRef {
            start_offset: start,
            end_offset: start + 4,
            s3_path: format!("benchcmp/compact-runtime/chunks/{start}.bin"),
            object_size: 4,
            ..Default::default()
        };
        assert!(
            cold_store
                .read_chunk_range(&old, old.start_offset, 4)
                .await
                .is_err()
        );
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stale_cold_flush_batch_after_delete_recreate_is_classified_for_cleanup() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(RuntimeConfig::new(2, 8), cold_store);
    let stream = BucketStreamId::new("benchcmp", "stale-cold-runtime");
    let placement = runtime.locate(&stream);
    create_stream(&runtime, &stream).await;
    append_bytes(&runtime, &stream, b"abcdefghijklmnopqr").await;
    let candidates = runtime
        .plan_next_cold_flush_batch(
            placement.raft_group_id,
            PlanGroupColdFlushRequest {
                min_hot_bytes: 18,
                max_flush_bytes: 18,
                max_batch_bytes: 18,
            },
            1,
        )
        .await
        .expect("plan candidate");
    assert_eq!(candidates.len(), 1);

    runtime
        .delete_stream(DeleteStreamRequest {
            stream_id: stream.clone(),
        })
        .await
        .expect("delete old stream");
    create_stream(&runtime, &stream).await;
    append_bytes(&runtime, &stream, b"abcdefghijklmnopq").await;

    let flushed = runtime
        .flush_cold_candidates_batch(candidates)
        .await
        .expect("stale candidate should be skipped");
    assert!(flushed.is_empty());
    let metrics = runtime.metrics().snapshot();
    assert_eq!(metrics.cold_flush_uploads, 1);
    assert_eq!(metrics.cold_flush_publishes, 0);
    assert_eq!(metrics.cold_orphan_cleanup_attempts, 0);
    assert_eq!(metrics.cold_orphan_cleanup_errors, 0);

    let read = runtime
        .read_stream(read_req(stream, 0, 32))
        .await
        .expect("read recreated stream");
    assert_eq!(read.payload, b"abcdefghijklmnopq");
    assert_eq!(read.next_offset, 17);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cold_write_admission_rejects_new_bytes_until_flush_catches_up() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(
        RuntimeConfig::new(2, 8).with_cold_max_hot_bytes_per_group(Some(4)),
        cold_store,
    );
    let stream = BucketStreamId::new("benchcmp", "cold-admission");
    create_stream(&runtime, &stream).await;
    append_bytes(&runtime, &stream, b"abcd").await;

    let err = runtime
        .append(AppendRequest::from_bytes(stream.clone(), b"e".to_vec()))
        .await
        .expect_err("append should be backpressured");
    match err {
        RuntimeError::GroupEngine {
            error:
                GroupEngineError::Infra(GroupInfraError::ColdBackpressure {
                    stream_id,
                    before_group_hot_bytes,
                    after_group_hot_bytes,
                    limit,
                    ..
                }),
            ..
        } => {
            assert_eq!(stream_id, stream);
            assert_eq!(before_group_hot_bytes, 4);
            assert_eq!(after_group_hot_bytes, 5);
            assert_eq!(limit, 4);
        }
        other => panic!("expected cold backpressure, got {other:?}"),
    }
    let metrics = runtime.metrics().snapshot();
    let group_index = usize::try_from(runtime.locate(&stream).raft_group_id.0).unwrap();
    assert_eq!(metrics.accepted_appends, 1);
    assert_eq!(metrics.cold_hot_bytes, 4);
    assert_eq!(metrics.per_group_cold_hot_bytes[group_index], 4);
    assert_eq!(metrics.cold_hot_group_bytes_max, 4);
    assert_eq!(metrics.cold_hot_stream_bytes_max, 4);
    assert_eq!(metrics.cold_backpressure_events, 1);
    assert_eq!(metrics.per_group_cold_backpressure_events[group_index], 1);
    assert_eq!(metrics.cold_backpressure_bytes, 1);

    runtime
        .flush_cold_once(PlanColdFlushRequest {
            stream_id: stream.clone(),
            min_hot_bytes: 4,
            max_flush_bytes: 4,
        })
        .await
        .expect("flush once")
        .expect("candidate flushed");
    let metrics_after_flush = runtime.metrics().snapshot();
    assert_eq!(metrics_after_flush.cold_hot_bytes, 0);
    assert_eq!(metrics_after_flush.cold_hot_group_bytes_max, 0);
    assert_eq!(
        metrics_after_flush.per_group_cold_hot_bytes_max[group_index], 4,
        "the diagnostic high-water mark remains monotonic"
    );

    append_bytes(&runtime, &stream, b"e").await;
    let read = runtime
        .read_stream(read_req(stream, 0, 5))
        .await
        .expect("read cold and hot");
    assert_eq!(read.payload, b"abcde");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cold_write_admission_allows_deduplicated_append_retry_at_hot_limit() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(
        RuntimeConfig::new(2, 8).with_cold_max_hot_bytes_per_group(Some(4)),
        cold_store,
    );
    let stream = BucketStreamId::new("benchcmp", "cold-admission-dedup-append");
    create_stream(&runtime, &stream).await;

    let mut first = AppendRequest::from_bytes(stream.clone(), b"abcd".to_vec());
    first.producer = Some(producer("writer", 0, 0));
    runtime.append(first).await.expect("append at hot limit");

    let mut retry = AppendRequest::from_bytes(stream.clone(), b"ignored".to_vec());
    retry.producer = Some(producer("writer", 0, 0));
    let retry = runtime
        .append(retry)
        .await
        .expect("deduplicated retry should bypass cold admission");
    assert!(retry.deduplicated);
    assert_eq!(retry.start_offset, 0);
    assert_eq!(retry.next_offset, 4);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cold_write_admission_allows_deduplicated_append_batch_retry_at_hot_limit() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(
        RuntimeConfig::new(2, 8).with_cold_max_hot_bytes_per_group(Some(4)),
        cold_store,
    );
    let stream = BucketStreamId::new("benchcmp", "cold-admission-dedup-batch");
    create_stream(&runtime, &stream).await;

    let mut first = AppendBatchRequest::new(stream.clone(), vec![
        b"a".to_vec(),
        b"b".to_vec(),
        b"c".to_vec(),
        b"d".to_vec(),
    ]);
    first.producer = Some(producer("writer", 0, 0));
    runtime
        .append_batch(first)
        .await
        .expect("batch at hot limit");

    let mut retry = AppendBatchRequest::new(stream.clone(), vec![b"ignored".to_vec()]);
    retry.producer = Some(producer("writer", 0, 0));
    let retry = runtime
        .append_batch(retry)
        .await
        .expect("deduplicated batch retry should bypass cold admission");
    assert!(
        retry
            .items
            .iter()
            .all(|item| item.as_ref().expect("deduplicated item").deduplicated)
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cold_write_admission_allows_existing_create_retry_at_hot_limit() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(
        RuntimeConfig::new(2, 8).with_cold_max_hot_bytes_per_group(Some(4)),
        cold_store,
    );
    let stream = BucketStreamId::new("benchcmp", "cold-admission-existing-create");
    let mut request = CreateStreamRequest::new(stream.clone(), DEFAULT_CONTENT_TYPE);
    request.initial_payload = Bytes::from_static(b"abcd");

    runtime
        .create_stream(request.clone())
        .await
        .expect("create at hot limit");
    let retry = runtime
        .create_stream(request)
        .await
        .expect("existing create should bypass cold admission");
    assert!(retry.already_exists);
    assert_eq!(retry.next_offset, 4);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn raft_uncommitted_admission_disabled_by_default_lets_writes_through() {
    // Disabled admission must not interfere with the existing accept path
    // (acceptance criterion 6: existing cold behaviour unchanged when the
    // raft admission is disabled).
    let runtime =
        spawn_in_memory(RuntimeConfig::new(2, 4).with_raft_max_uncommitted_bytes_per_group(None));
    let stream = BucketStreamId::new("benchcmp", "raft-uncommitted-disabled");
    create_stream(&runtime, &stream).await;
    for _ in 0..4 {
        append_bytes(&runtime, &stream, b"payload").await;
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn raft_uncommitted_admission_rejects_when_incoming_would_exceed_limit() {
    // With a 4-byte cap and a 5-byte append on an empty stream, the
    // CoreWorker-level admission rejects the request before reaching the
    // actor since `current (0) + incoming (5) > limit (4)`.
    let runtime = spawn_in_memory(
        RuntimeConfig::new(2, 4).with_raft_max_uncommitted_bytes_per_group(Some(4)),
    );
    let stream = BucketStreamId::new("benchcmp", "raft-uncommitted-trip");
    create_stream(&runtime, &stream).await;

    let err = runtime
        .append(AppendRequest::from_bytes(stream.clone(), vec![b'x'; 5]))
        .await
        .expect_err("oversized append should trip raft uncommitted admission");
    match err {
        RuntimeError::GroupEngine {
            error:
                GroupEngineError::Infra(GroupInfraError::RaftUncommittedBackpressure {
                    current,
                    incoming,
                    limit,
                    ..
                }),
            ..
        } => {
            assert_eq!(current, 0);
            assert_eq!(incoming, 5);
            assert_eq!(limit, 4);
        }
        other => panic!("expected raft uncommitted backpressure, got {other:?}"),
    }

    // A within-budget append still succeeds.
    append_bytes(&runtime, &stream, b"abcd").await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cold_write_admission_rejects_append_batch_without_partial_mutation() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(
        RuntimeConfig::new(2, 8).with_cold_max_hot_bytes_per_group(Some(4)),
        cold_store,
    );
    let stream = BucketStreamId::new("benchcmp", "cold-admission-batch");
    create_stream(&runtime, &stream).await;
    append_bytes(&runtime, &stream, b"abc").await;

    let err = runtime
        .append_batch(AppendBatchRequest::new(stream.clone(), vec![
            b"d".to_vec(),
            b"e".to_vec(),
        ]))
        .await
        .expect_err("batch should be backpressured");
    match err {
        RuntimeError::GroupEngine {
            error:
                GroupEngineError::Infra(GroupInfraError::ColdBackpressure {
                    stream_id,
                    before_group_hot_bytes,
                    after_group_hot_bytes,
                    limit,
                    ..
                }),
            ..
        } => {
            assert_eq!(stream_id, stream);
            assert_eq!(before_group_hot_bytes, 3);
            assert_eq!(after_group_hot_bytes, 5);
            assert_eq!(limit, 4);
        }
        other => panic!("expected cold backpressure, got {other:?}"),
    }
    let read = runtime
        .read_stream(read_req(stream.clone(), 0, 8))
        .await
        .expect("read");
    assert_eq!(read.payload, b"abc");
    let metrics = runtime.metrics().snapshot();
    assert_eq!(metrics.accepted_appends, 1);
    assert_eq!(metrics.cold_backpressure_events, 1);
    assert_eq!(metrics.cold_backpressure_bytes, 2);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cold_write_admission_does_not_preempt_non_local_write_engine() {
    let factory = RecordingFactory::without_local_writes().with_cold_hot_bytes(8);
    let runtime = ShardRuntime::spawn_with_engine_factory(
        RuntimeConfig::new(2, 8).with_cold_max_hot_bytes_per_group(Some(4)),
        factory,
    )
    .expect("spawn runtime");
    let stream = BucketStreamId::new("benchcmp", "cold-admission-non-local");

    runtime
        .create_stream(CreateStreamRequest::new(
            stream.clone(),
            DEFAULT_CONTENT_TYPE,
        ))
        .await
        .expect("create reaches non-local engine");
    assert_eq!(runtime.metrics().snapshot().cold_hot_bytes, 8);

    runtime
        .append(AppendRequest::from_bytes(stream, b"x".to_vec()))
        .await
        .expect("append reaches non-local engine despite local cold backlog");

    let metrics = runtime.metrics().snapshot();
    assert_eq!(metrics.accepted_appends, 1);
    assert_eq!(metrics.cold_backpressure_events, 0);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn flush_cold_group_once_selects_stream_inside_owner_group() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(RuntimeConfig::new(2, 8), cold_store);
    let group_id = RaftGroupId(3);
    let stream = stream_on_group(&runtime, group_id, "cold-group");
    create_stream(&runtime, &stream).await;
    append_bytes(&runtime, &stream, b"abcdef").await;

    let flushed = runtime
        .flush_cold_group_once(group_id, PlanGroupColdFlushRequest {
            min_hot_bytes: 4,
            max_flush_bytes: 4,
            max_batch_bytes: 4,
        })
        .await
        .expect("flush group")
        .expect("candidate flushed");
    assert_eq!(flushed.hot_start_offset, 4);

    let read = runtime
        .read_stream(read_req(stream, 0, 6))
        .await
        .expect("read cold and hot");
    assert_eq!(read.payload, b"abcdef");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn flush_cold_all_groups_once_bounded_flushes_multiple_groups() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(RuntimeConfig::new(2, 8), cold_store);
    let first = stream_on_group(&runtime, RaftGroupId(1), "cold-bounded-a");
    let second = stream_on_group(&runtime, RaftGroupId(6), "cold-bounded-b");
    for stream in [&first, &second] {
        create_stream(&runtime, stream).await;
        append_bytes(&runtime, stream, b"abcdef").await;
    }

    let flushed = runtime
        .flush_cold_all_groups_once_bounded(
            PlanGroupColdFlushRequest {
                min_hot_bytes: 4,
                max_flush_bytes: 4,
                max_batch_bytes: 8,
            },
            2,
        )
        .await
        .expect("flush all bounded");
    assert_eq!(flushed, 4);
    let metrics = runtime.metrics().snapshot();
    assert_eq!(metrics.cold_flush_uploads, 2);
    assert_eq!(metrics.cold_flush_upload_bytes, 12);
    assert_eq!(metrics.cold_pack_uploads, 2);
    assert_eq!(metrics.cold_pack_slices, 4);
    assert_eq!(metrics.cold_flush_publishes, 4);
    assert_eq!(metrics.cold_flush_publish_bytes, 12);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn repeated_cold_flush_keeps_hot_bytes_bounded_while_writes_continue() {
    let cold_store = Arc::new(memory_cold_store());
    let runtime = spawn_with_cold_store(
        RuntimeConfig::new(2, 8).with_cold_max_hot_bytes_per_group(Some(16)),
        cold_store,
    );
    let streams = [
        stream_on_group(&runtime, RaftGroupId(0), "cold-steady-a"),
        stream_on_group(&runtime, RaftGroupId(3), "cold-steady-b"),
        stream_on_group(&runtime, RaftGroupId(5), "cold-steady-c"),
        stream_on_group(&runtime, RaftGroupId(7), "cold-steady-d"),
    ];
    for stream in &streams {
        create_stream(&runtime, stream).await;
    }

    let mut expected = Vec::new();
    for round in 0..8u8 {
        let payload = vec![b'a' + round; 4];
        expected.extend_from_slice(&payload);
        for stream in &streams {
            runtime
                .append(AppendRequest::from_bytes(stream.clone(), payload.clone()))
                .await
                .expect("append while cold worker keeps up");
        }

        let metrics_before_flush = runtime.metrics().snapshot();
        assert!(
            metrics_before_flush.cold_hot_bytes <= 64,
            "hot bytes should stay within one unflushed batch per group before flush: {}",
            metrics_before_flush.cold_hot_bytes
        );

        let flushed = runtime
            .flush_cold_all_groups_once_bounded(
                PlanGroupColdFlushRequest {
                    min_hot_bytes: 4,
                    max_flush_bytes: 4,
                    max_batch_bytes: 4,
                },
                streams.len(),
            )
            .await
            .expect("flush all bounded");
        assert_eq!(flushed, streams.len());
        let metrics_after_flush = runtime.metrics().snapshot();
        assert_eq!(
            metrics_after_flush.cold_hot_bytes, 0,
            "all newly appended bytes should be offloaded after round {round}"
        );
        assert_eq!(
            metrics_after_flush.cold_flush_uploads,
            u64::try_from((usize::from(round) + 1) * streams.len()).expect("count fits u64")
        );
        assert_eq!(metrics_after_flush.cold_orphan_cleanup_attempts, 0);
        assert_eq!(metrics_after_flush.cold_backpressure_events, 0);
    }

    for stream in streams {
        let read = runtime
            .read_stream(read_req(stream, 0, expected.len()))
            .await
            .expect("read cold-backed stream");
        assert_eq!(read.payload, expected);
        assert_eq!(read.next_offset, u64::try_from(expected.len()).unwrap());
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn wait_read_stream_completes_after_owner_append() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "wait-read");
    create_stream(&runtime, &stream).await;

    let wait = {
        let runtime = runtime.clone();
        let stream = stream.clone();
        tokio::spawn(async move {
            runtime
                .wait_read_stream(read_req(stream, 0, 16))
                .await
                .expect("wait read")
        })
    };
    tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    append_bytes(&runtime, &stream, b"hello").await;

    let read = tokio::time::timeout(std::time::Duration::from_secs(1), wait)
        .await
        .expect("wait read timeout")
        .expect("wait task");
    assert_eq!(read.payload, b"hello");
    assert_eq!(read.next_offset, 5);
    assert!(read.up_to_date);
    assert!(!read.closed);
    assert_eq!(runtime.metrics().snapshot().live_read_waiters, 0);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn wait_read_stream_completes_on_close_at_tail() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "wait-close");
    create_stream(&runtime, &stream).await;

    let wait = {
        let runtime = runtime.clone();
        let stream = stream.clone();
        tokio::spawn(async move {
            runtime
                .wait_read_stream(read_req(stream, 0, 16))
                .await
                .expect("wait read")
        })
    };
    tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    runtime
        .close_stream(CloseStreamRequest {
            stream_id: stream,
            stream_seq: None,
            producer: None,
            now_ms: 0,
        })
        .await
        .expect("close stream");

    let read = tokio::time::timeout(std::time::Duration::from_secs(1), wait)
        .await
        .expect("wait read timeout")
        .expect("wait task");
    assert!(read.payload.is_empty());
    assert_eq!(read.next_offset, 0);
    assert!(read.up_to_date);
    assert!(read.closed);
    assert_eq!(runtime.metrics().snapshot().live_read_waiters, 0);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn canceled_wait_read_stream_removes_owner_waiter() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "wait-cancel");
    create_stream(&runtime, &stream).await;

    let wait = {
        let runtime = runtime.clone();
        let stream = stream.clone();
        tokio::spawn(async move { runtime.wait_read_stream(read_req(stream, 0, 16)).await })
    };
    wait_for_live_waiters(&runtime, 1).await;
    wait.abort();
    let _ = wait.await;
    wait_for_live_waiters(&runtime, 0).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn live_read_waiter_limit_rejects_excess_waiters_on_owner_core() {
    let runtime =
        ShardRuntime::spawn(RuntimeConfig::new(1, 1).with_live_read_max_waiters_per_core(Some(1)))
            .expect("spawn runtime");
    let stream = BucketStreamId::new("benchcmp", "wait-limit");
    create_stream(&runtime, &stream).await;

    let first = {
        let runtime = runtime.clone();
        let stream = stream.clone();
        tokio::spawn(async move { runtime.wait_read_stream(read_req(stream, 0, 16)).await })
    };
    wait_for_live_waiters(&runtime, 1).await;

    let err = runtime
        .wait_read_stream(read_req(stream.clone(), 0, 16))
        .await
        .expect_err("second waiter should hit owner-core limit");
    assert_eq!(err, RuntimeError::LiveReadBackpressure {
        core_id: CoreId(0),
        current_waiters: 1,
        limit: 1,
    });
    let snapshot = runtime.metrics().snapshot();
    assert_eq!(snapshot.live_read_waiters, 1);
    assert_eq!(snapshot.live_read_backpressure_events, 1);
    assert_eq!(snapshot.per_core_live_read_backpressure_events, vec![1]);

    first.abort();
    let _ = first.await;
    wait_for_live_waiters(&runtime, 0).await;
}

#[test]
fn cancel_read_watcher_removes_group_local_waiter() {
    let stream = BucketStreamId::new("benchcmp", "watcher-cancel-local");
    let mut read_watchers = ReadWatchers::new();
    let (first_tx, _first_rx) = oneshot::channel();
    let (second_tx, _second_rx) = oneshot::channel();
    read_watchers.insert(stream.clone(), vec![
        ReadWatcher {
            waiter_id: 1,
            request: read_req(stream.clone(), 0, 16),
            response_tx: first_tx,
        },
        ReadWatcher {
            waiter_id: 2,
            request: read_req(stream.clone(), 0, 16),
            response_tx: second_tx,
        },
    ]);

    let metrics = Arc::new(RuntimeMetricsInner::new(1, 1));
    metrics.record_read_watchers_added(CoreId(0), 2);
    CoreWorker::cancel_read_watcher(
        &mut read_watchers,
        metrics.clone(),
        CoreId(0),
        stream.clone(),
        1,
    );

    let watcher_ids = read_watchers
        .get(&stream)
        .expect("one watcher remains")
        .iter()
        .map(|watcher| watcher.waiter_id)
        .collect::<Vec<_>>();
    assert_eq!(watcher_ids, vec![2]);
    assert_eq!(metrics.per_core_live_read_waiters[0].load_relaxed(), 1);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn notify_read_watchers_shares_identical_reads_across_watchers() {
    let factory = BlockingReadFactory::default();
    let runtime = ShardRuntime::spawn_with_engine_factory(test_config(1, 1, 8), factory.clone())
        .expect("spawn runtime");
    let stream = BucketStreamId::new("benchcmp", "watcher-shared-read");
    let placement = runtime.locate(&stream);
    let request = read_req(stream.clone(), 0, 16);
    let mut read_watchers = ReadWatchers::new();
    let (first_tx, _first_rx) = oneshot::channel();
    let (second_tx, _second_rx) = oneshot::channel();
    read_watchers.insert(stream.clone(), vec![
        ReadWatcher {
            waiter_id: 1,
            request: request.clone(),
            response_tx: first_tx,
        },
        ReadWatcher {
            waiter_id: 2,
            request,
            response_tx: second_tx,
        },
    ]);

    let metrics = Arc::new(RuntimeMetricsInner::new(1, 1));
    let mut engine = factory
        .create(placement, GroupEngineMetrics {
            inner: metrics.clone(),
        })
        .await
        .expect("create engine");
    let notify = {
        let stream = stream.clone();
        tokio::spawn(async move {
            CoreWorker::notify_read_watchers(
                &mut engine,
                metrics,
                Arc::new(Semaphore::new(8)),
                &mut read_watchers,
                &stream,
                placement,
            )
            .await;
            read_watchers
        })
    };
    tokio::time::timeout(
        std::time::Duration::from_secs(1),
        factory.entered.notified(),
    )
    .await
    .expect("notify issued one grouped read");
    factory.release.notify_one();
    let read_watchers = tokio::time::timeout(std::time::Duration::from_secs(1), notify)
        .await
        .expect("notify should finish after one read")
        .expect("notify task");

    let watcher_ids = read_watchers
        .get(&stream)
        .expect("pending watchers reinserted")
        .iter()
        .map(|watcher| watcher.waiter_id)
        .collect::<Vec<_>>();
    assert_eq!(watcher_ids, vec![1, 2]);
    assert_eq!(factory.read_count.load(Ordering::Relaxed), 1);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn close_stream_allows_close_only_and_rejects_later_appends() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "close-only");
    let placement = runtime.locate(&stream);
    create_stream(&runtime, &stream).await;

    let closed = runtime
        .close_stream(CloseStreamRequest {
            stream_id: stream.clone(),
            stream_seq: None,
            producer: None,
            now_ms: 0,
        })
        .await
        .expect("close stream");
    assert_eq!(closed.placement, placement);
    assert_eq!(closed.next_offset, 0);

    let err = runtime
        .append(AppendRequest::new(stream.clone(), 1))
        .await
        .expect_err("append after close rejected");
    match err {
        RuntimeError::GroupEngine { error, .. } => {
            let message = error.message();
            assert!(message.contains("StreamClosed"), "message={message}");
        }
        other => panic!("expected group engine error, got {other:?}"),
    }

    let head = runtime
        .head_stream(HeadStreamRequest {
            stream_id: stream,
            now_ms: 0,
        })
        .await
        .expect("head stream");
    assert_eq!(head.tail_offset, 0);
    assert!(head.closed);
    assert_eq!(runtime.metrics().snapshot().accepted_appends, 0);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn delete_stream_removes_state_on_owner_group() {
    let runtime = runtime(2, 8);
    let stream = BucketStreamId::new("benchcmp", "delete-stream");
    let placement = runtime.locate(&stream);
    create_stream(&runtime, &stream).await;
    append_bytes(&runtime, &stream, b"payload").await;

    let deleted = runtime
        .delete_stream(DeleteStreamRequest {
            stream_id: stream.clone(),
        })
        .await
        .expect("delete stream");
    assert_eq!(deleted.placement, placement);

    let err = runtime
        .head_stream(HeadStreamRequest {
            stream_id: stream.clone(),
            now_ms: 0,
        })
        .await
        .expect_err("head after delete rejected");
    match err {
        RuntimeError::GroupEngine { error, .. } => {
            let message = error.message();
            assert!(message.contains("StreamNotFound"), "message={message}");
        }
        other => panic!("expected group engine error, got {other:?}"),
    }

    let err = runtime
        .append(AppendRequest::new(stream, 1))
        .await
        .expect_err("append after delete rejected");
    match err {
        RuntimeError::GroupEngine { error, .. } => {
            let message = error.message();
            assert!(message.contains("StreamNotFound"), "message={message}");
        }
        other => panic!("expected group engine error, got {other:?}"),
    }
}

#[cfg(not(madsim))]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn thread_per_core_runtime_reaches_all_configured_cores() {
    let mut config = RuntimeConfig::new(4, 32);
    config.mailbox_capacity = 128;
    assert_eq!(config.threading, RuntimeThreading::ThreadPerCore);
    let runtime = ShardRuntime::spawn(config).expect("spawn runtime");

    let mut tasks = Vec::new();
    for index in 0..1024 {
        let runtime = runtime.clone();
        tasks.push(tokio::spawn(async move {
            let stream = BucketStreamId::new("benchcmp", format!("thread-core-{index}"));
            create_stream(&runtime, &stream).await;
            runtime
                .append(AppendRequest::new(stream, 1))
                .await
                .expect("append");
        }));
    }

    for task in tasks {
        task.await.expect("task");
    }

    let snapshot = runtime.metrics().snapshot();
    assert_eq!(snapshot.accepted_appends, 1024);
    assert!(snapshot.per_core_appends.iter().all(|value| *value > 0));
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn custom_group_engine_is_created_once_per_touched_group_on_owner_core() {
    let factory = RecordingFactory::default();
    let runtime = ShardRuntime::spawn_with_engine_factory(test_config(4, 32, 128), factory.clone())
        .expect("spawn runtime");

    let mut touched_groups = HashSet::new();
    for index in 0..4096 {
        let stream = BucketStreamId::new("benchcmp", format!("engine-{index}"));
        let placement = runtime.locate(&stream);
        runtime
            .create_stream(CreateStreamRequest::new(stream, DEFAULT_CONTENT_TYPE))
            .await
            .expect("create stream");
        touched_groups.insert(placement.raft_group_id);
        if touched_groups.len() == 16 {
            break;
        }
    }

    let created = factory.created();
    let created_groups = created
        .iter()
        .map(|placement| placement.raft_group_id)
        .collect::<HashSet<_>>();
    assert_eq!(created_groups, touched_groups);
    for placement in created {
        assert_eq!(
            u32::from(placement.core_id.0),
            placement.raft_group_id.0 % 4
        );
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn background_cold_flush_skips_groups_that_cannot_accept_local_writes() {
    let factory = RecordingFactory::without_local_writes();
    let cold_store = Arc::new(memory_cold_store());
    let runtime = ShardRuntime::spawn_with_engine_factory_and_cold_store(
        test_config(2, 4, 128),
        factory.clone(),
        Some(cold_store),
    )
    .expect("spawn runtime");

    let flushed = runtime
        .flush_cold_all_groups_once_bounded(
            PlanGroupColdFlushRequest {
                min_hot_bytes: 1,
                max_flush_bytes: 1,
                max_batch_bytes: 4,
            },
            4,
        )
        .await
        .expect("flush all groups");

    assert_eq!(flushed, 0);
    assert_eq!(factory.created().len(), 4);
    let metrics = runtime.metrics().snapshot();
    assert_eq!(metrics.cold_flush_uploads, 0);
    assert_eq!(metrics.cold_flush_publishes, 0);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn warm_group_instantiates_engine_on_owner_core_without_stream_mutation() {
    let factory = RecordingFactory::default();
    let runtime = ShardRuntime::spawn_with_engine_factory(test_config(2, 4, 128), factory.clone())
        .expect("spawn runtime");

    let warmed = runtime
        .warm_group(RaftGroupId(3))
        .await
        .expect("warm group");
    assert_eq!(warmed.core_id, CoreId(1));
    assert_eq!(warmed.raft_group_id, RaftGroupId(3));

    runtime
        .warm_group(RaftGroupId(3))
        .await
        .expect("second warm is idempotent");

    let created = factory.created();
    assert_eq!(created, vec![warmed]);

    runtime.warm_all_groups().await.expect("warm all groups");
    let created_groups = factory
        .created()
        .into_iter()
        .map(|placement| placement.raft_group_id)
        .collect::<HashSet<_>>();
    assert_eq!(
        created_groups,
        [
            RaftGroupId(0),
            RaftGroupId(1),
            RaftGroupId(2),
            RaftGroupId(3)
        ]
        .into_iter()
        .collect()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn warm_all_groups_warms_owner_cores_concurrently() {
    let factory = BlockingWarmFactory::default();
    let runtime = ShardRuntime::spawn_with_engine_factory(test_config(2, 2, 128), factory.clone())
        .expect("spawn runtime");
    let blocked_entered = factory.blocked_entered.notified();
    let other_created = factory.other_created.notified();
    let warm = tokio::spawn(async move { runtime.warm_all_groups().await });

    tokio::time::timeout(Duration::from_secs(1), blocked_entered)
        .await
        .expect("group zero started warming");
    tokio::time::timeout(Duration::from_secs(1), other_created)
        .await
        .expect("other owner core should warm while group zero is blocked");

    factory.release.notify_one();
    warm.await.expect("warm task").expect("warm all groups");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn warm_all_groups_warms_groups_within_one_core_concurrently() {
    let factory = BlockingWarmFactory::default();
    let runtime = ShardRuntime::spawn_with_engine_factory(test_config(1, 2, 128), factory.clone())
        .expect("spawn runtime");
    let blocked_entered = factory.blocked_entered.notified();
    let other_created = factory.other_created.notified();
    let warm = tokio::spawn(async move { runtime.warm_all_groups().await });

    tokio::time::timeout(Duration::from_secs(1), blocked_entered)
        .await
        .expect("group zero started warming");
    tokio::time::timeout(Duration::from_secs(1), other_created)
        .await
        .expect("another group on the same core should warm concurrently");

    factory.release.notify_one();
    warm.await.expect("warm task").expect("warm all groups");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn core_worker_dispatches_other_groups_while_one_group_waits() {
    let factory = BlockingFirstCreateEngineFactory::default();
    let runtime = ShardRuntime::spawn_with_engine_factory(test_config(1, 2, 128), factory.clone())
        .expect("spawn runtime");

    let blocked_stream = stream_on_group(&runtime, RaftGroupId(0), "blocked-group");
    let free_stream = stream_on_group(&runtime, RaftGroupId(1), "free-group");
    let entered_wait = factory.entered.notified();
    let blocked_runtime = runtime.clone();
    let blocked =
        tokio::spawn(async move { create_stream(&blocked_runtime, &blocked_stream).await });

    tokio::time::timeout(std::time::Duration::from_secs(1), entered_wait)
        .await
        .expect("first group entered blocking create");

    let completed = tokio::time::timeout(
        std::time::Duration::from_secs(1),
        create_stream(&runtime, &free_stream),
    )
    .await
    .expect("other group should complete while first group is blocked");
    assert_eq!(completed.placement.raft_group_id, RaftGroupId(1));

    factory.release.notify_one();
    blocked.await.expect("blocked task");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn runtime_read_uses_group_read_parts_fast_path() {
    let factory = BlockingReadFactory::default();
    let runtime = ShardRuntime::spawn_with_engine_factory(test_config(1, 1, 128), factory.clone())
        .expect("spawn runtime");
    let stream = BucketStreamId::new("benchcmp", "read-offload");
    create_stream(&runtime, &stream).await;

    let read = tokio::time::timeout(
        std::time::Duration::from_secs(1),
        runtime.read_stream(read_req(stream.clone(), 0, 16)),
    )
    .await
    .expect("runtime read should not use blocking legacy read_stream")
    .expect("read stream");
    assert_eq!(read.placement.raft_group_id, RaftGroupId(0));
    assert_eq!(factory.read_count.load(Ordering::Relaxed), 1);

    let head = runtime
        .head_stream(HeadStreamRequest {
            stream_id: stream,
            now_ms: 0,
        })
        .await
        .expect("head stream");
    assert_eq!(head.placement.raft_group_id, RaftGroupId(0));
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn read_materialization_is_bounded_without_blocking_group_actor() {
    let factory = BlockingReadFactory::block_materialization();
    let mut config = RuntimeConfig::new(1, 1);
    config.mailbox_capacity = 1;
    config.threading = RuntimeThreading::HostedTokio;
    let runtime =
        ShardRuntime::spawn_with_engine_factory(config, factory.clone()).expect("spawn runtime");
    let first_stream = BucketStreamId::new("benchcmp", "materialize-bound-1");
    let second_stream = BucketStreamId::new("benchcmp", "materialize-bound-2");
    create_stream(&runtime, &first_stream).await;
    create_stream(&runtime, &second_stream).await;

    let first_runtime = runtime.clone();
    let first_stream_for_read = first_stream.clone();
    let first_read = tokio::spawn(async move {
        first_runtime
            .read_stream(read_req(first_stream_for_read, 0, 16))
            .await
    });
    tokio::time::timeout(
        std::time::Duration::from_secs(1),
        factory.materialized.notified(),
    )
    .await
    .expect("first materialization acquired the only permit");

    let second_runtime = runtime.clone();
    let second_stream_for_read = second_stream.clone();
    let second_read = tokio::spawn(async move {
        second_runtime
            .read_stream(read_req(second_stream_for_read, 0, 16))
            .await
    });

    let head = tokio::time::timeout(
        std::time::Duration::from_secs(1),
        runtime.head_stream(HeadStreamRequest {
            stream_id: first_stream,
            now_ms: 0,
        }),
    )
    .await
    .expect("group actor should keep serving metadata while materialization waits")
    .expect("head stream");
    assert_eq!(head.placement.raft_group_id, RaftGroupId(0));
    assert!(!second_read.is_finished());

    factory.release.notify_one();
    let first = first_read
        .await
        .expect("first read task")
        .expect("first read");
    assert_eq!(first.payload, b"ready");
    tokio::time::timeout(
        std::time::Duration::from_secs(1),
        factory.materialized.notified(),
    )
    .await
    .expect("second materialization acquired permit after first released it");
    factory.release.notify_one();
    let second = second_read
        .await
        .expect("second read task")
        .expect("second read");
    assert_eq!(second.payload, b"ready");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn group_engine_errors_include_group_context_and_do_not_record_success_metrics() {
    let runtime = ShardRuntime::spawn_with_engine_factory(test_config(2, 8, 128), FailingFactory)
        .expect("spawn runtime");

    let stream = BucketStreamId::new("benchcmp", "failing-stream");
    let placement = runtime.locate(&stream);
    let err = runtime
        .append(AppendRequest::new(stream, 1))
        .await
        .expect_err("engine failure");

    assert_eq!(err, RuntimeError::GroupEngine {
        core_id: placement.core_id,
        raft_group_id: placement.raft_group_id,
        error: GroupEngineError::new("proposal rejected"),
    });
    assert_eq!(runtime.metrics().snapshot().accepted_appends, 0);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn mailbox_full_events_record_owner_core_backpressure() {
    let factory = BlockingOnceFactory::default();
    let runtime = ShardRuntime::spawn_with_engine_factory(test_config(1, 1, 1), factory.clone())
        .expect("spawn runtime");

    let entered = factory.entered.clone();
    let entered_wait = entered.notified();
    let first_runtime = runtime.clone();
    let first = tokio::spawn(async move {
        create_stream(
            &first_runtime,
            &BucketStreamId::new("benchcmp", "backpressure-1"),
        )
        .await
    });
    tokio::time::timeout(std::time::Duration::from_secs(1), entered_wait)
        .await
        .expect("first create entered blocking engine factory");

    let second_runtime = runtime.clone();
    let second = tokio::spawn(async move {
        create_stream(
            &second_runtime,
            &BucketStreamId::new("benchcmp", "backpressure-2"),
        )
        .await
    });
    wait_for_mailbox_depth(&runtime, 0, 1).await;

    let third_runtime = runtime.clone();
    let third = tokio::spawn(async move {
        create_stream(
            &third_runtime,
            &BucketStreamId::new("benchcmp", "backpressure-3"),
        )
        .await
    });
    wait_for_mailbox_full_events(&runtime, 1).await;
    assert_eq!(
        runtime.metrics().snapshot().per_core_mailbox_full_events[0],
        1
    );

    factory.release.notify_one();
    first.await.expect("first task");
    second.await.expect("second task");
    third.await.expect("third task");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn group_mailbox_full_events_record_inner_actor_backpressure() {
    let factory = BlockingFirstCreateEngineFactory::default();
    let runtime = ShardRuntime::spawn_with_engine_factory(test_config(1, 1, 1), factory.clone())
        .expect("spawn runtime");

    let first_runtime = runtime.clone();
    let first = tokio::spawn(async move {
        create_stream(
            &first_runtime,
            &BucketStreamId::new("benchcmp", "group-backpressure-1"),
        )
        .await
    });
    tokio::time::timeout(
        std::time::Duration::from_secs(1),
        factory.entered.notified(),
    )
    .await
    .expect("first append entered blocking group engine");

    let second_runtime = runtime.clone();
    let second = tokio::spawn(async move {
        create_stream(
            &second_runtime,
            &BucketStreamId::new("benchcmp", "group-backpressure-2"),
        )
        .await
    });
    for _ in 0..100 {
        if runtime.metrics().snapshot().group_mailbox_depth == 1 {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }

    let third_runtime = runtime.clone();
    let third = tokio::spawn(async move {
        create_stream(
            &third_runtime,
            &BucketStreamId::new("benchcmp", "group-backpressure-3"),
        )
        .await
    });
    wait_for_group_mailbox_full_events(&runtime, 1).await;
    assert_eq!(
        runtime
            .metrics()
            .snapshot()
            .per_group_group_mailbox_full_events[0],
        1
    );

    factory.release.notify_one();
    first.await.expect("first task");
    second.await.expect("second task");
    third.await.expect("third task");
}

#[derive(Debug, Clone)]
struct RecordingFactory {
    created: Arc<Mutex<Vec<ShardPlacement>>>,
    accepts_local_writes: bool,
    cold_hot_bytes: u64,
}

impl Default for RecordingFactory {
    fn default() -> Self {
        Self {
            created: Arc::default(),
            accepts_local_writes: true,
            cold_hot_bytes: 0,
        }
    }
}

impl RecordingFactory {
    fn without_local_writes() -> Self {
        Self {
            accepts_local_writes: false,
            ..Self::default()
        }
    }

    fn with_cold_hot_bytes(mut self, bytes: u64) -> Self {
        self.cold_hot_bytes = bytes;
        self
    }

    fn created(&self) -> Vec<ShardPlacement> {
        self.created.lock().expect("lock created groups").clone()
    }
}

impl GroupEngineFactory for RecordingFactory {
    fn create<'a>(
        &'a self,
        placement: ShardPlacement,
        _metrics: GroupEngineMetrics,
    ) -> GroupEngineCreateFuture<'a> {
        Box::pin(async move {
            self.created
                .lock()
                .expect("lock created groups")
                .push(placement);
            let engine: Box<dyn GroupEngine> = Box::new(RecordingEngine {
                placement,
                commit_index: 0,
                accepts_local_writes: self.accepts_local_writes,
                cold_hot_bytes: self.cold_hot_bytes,
            });
            Ok(engine)
        })
    }
}

struct RecordingEngine {
    placement: ShardPlacement,
    commit_index: u64,
    accepts_local_writes: bool,
    cold_hot_bytes: u64,
}

#[derive(Debug, Clone)]
struct BlockingWarmFactory {
    blocked_entered: Arc<Notify>,
    other_created: Arc<Notify>,
    release: Arc<Notify>,
}

impl Default for BlockingWarmFactory {
    fn default() -> Self {
        Self {
            blocked_entered: Arc::new(Notify::new()),
            other_created: Arc::new(Notify::new()),
            release: Arc::new(Notify::new()),
        }
    }
}

impl GroupEngineFactory for BlockingWarmFactory {
    fn create<'a>(
        &'a self,
        placement: ShardPlacement,
        _metrics: GroupEngineMetrics,
    ) -> GroupEngineCreateFuture<'a> {
        Box::pin(async move {
            if placement.raft_group_id == RaftGroupId(0) {
                self.blocked_entered.notify_one();
                self.release.notified().await;
            } else {
                self.other_created.notify_one();
            }
            let engine: Box<dyn GroupEngine> = Box::new(RecordingEngine {
                placement,
                commit_index: 0,
                accepts_local_writes: true,
                cold_hot_bytes: 0,
            });
            Ok(engine)
        })
    }
}

#[derive(Clone)]
struct BlockingReadFactory {
    entered: Arc<Notify>,
    materialized: Arc<Notify>,
    release: Arc<Notify>,
    read_count: Arc<AtomicU64>,
    block_parts: bool,
}

impl Default for BlockingReadFactory {
    fn default() -> Self {
        Self {
            entered: Arc::new(Notify::new()),
            materialized: Arc::new(Notify::new()),
            release: Arc::new(Notify::new()),
            read_count: Arc::new(AtomicU64::new(0)),
            block_parts: false,
        }
    }
}

impl BlockingReadFactory {
    fn block_materialization() -> Self {
        Self {
            block_parts: true,
            ..Self::default()
        }
    }
}

impl GroupEngineFactory for BlockingReadFactory {
    fn create<'a>(
        &'a self,
        placement: ShardPlacement,
        _metrics: GroupEngineMetrics,
    ) -> GroupEngineCreateFuture<'a> {
        Box::pin(async move {
            let engine: Box<dyn GroupEngine> = Box::new(BlockingReadEngine {
                inner: InMemoryGroupEngine::default(),
                placement,
                entered: self.entered.clone(),
                materialized: self.materialized.clone(),
                release: self.release.clone(),
                read_count: self.read_count.clone(),
                block_parts: self.block_parts,
            });
            Ok(engine)
        })
    }
}

struct BlockingReadEngine {
    inner: InMemoryGroupEngine,
    placement: ShardPlacement,
    entered: Arc<Notify>,
    materialized: Arc<Notify>,
    release: Arc<Notify>,
    read_count: Arc<AtomicU64>,
    block_parts: bool,
}

impl GroupEngine for BlockingReadEngine {
    fn create_stream<'a>(
        &'a mut self,
        request: CreateStreamRequest,
        placement: ShardPlacement,
        admission: ColdWriteAdmission,
    ) -> GroupCreateStreamFuture<'a> {
        self.inner.create_stream(request, placement, admission)
    }

    fn head_stream<'a>(
        &'a mut self,
        request: HeadStreamRequest,
        placement: ShardPlacement,
    ) -> GroupHeadStreamFuture<'a> {
        self.inner.head_stream(request, placement)
    }

    fn bucket_usage<'a>(&'a mut self, placement: ShardPlacement) -> GroupBucketUsageFuture<'a> {
        self.inner.bucket_usage(placement)
    }

    fn read_stream<'a>(
        &'a mut self,
        request: ReadStreamRequest,
        placement: ShardPlacement,
    ) -> GroupReadStreamFuture<'a> {
        let entered = self.entered.clone();
        let release = self.release.clone();
        let read_count = self.read_count.clone();
        Box::pin(async move {
            assert_eq!(placement, self.placement);
            read_count.fetch_add(1, Ordering::Relaxed);
            entered.notify_one();
            release.notified().await;
            Ok(ReadStreamResponse {
                placement,
                offset: request.offset,
                next_offset: request.offset,
                content_type: DEFAULT_CONTENT_TYPE.to_owned(),
                payload: Vec::new(),
                up_to_date: true,
                closed: false,
                retained_record_range: None,
                record_range: None,
            })
        })
    }

    fn read_stream_parts<'a>(
        &'a mut self,
        request: ReadStreamRequest,
        placement: ShardPlacement,
    ) -> GroupReadStreamPartsFuture<'a> {
        let entered = self.entered.clone();
        let read_count = self.read_count.clone();
        Box::pin(async move {
            assert_eq!(placement, self.placement);
            read_count.fetch_add(1, Ordering::Relaxed);
            entered.notify_one();
            if self.block_parts {
                return Ok(GroupReadStreamParts {
                    placement,
                    offset: request.offset,
                    next_offset: request.offset
                        + u64::try_from(b"ready".len()).expect("payload len fits u64"),
                    content_type: DEFAULT_CONTENT_TYPE.to_owned(),
                    up_to_date: true,
                    closed: false,
                    retained_record_range: None,
                    record_range: None,
                    body: GroupReadStreamBody::Blocking {
                        entered: self.entered.clone(),
                        materialized: self.materialized.clone(),
                        release: self.release.clone(),
                        payload: b"ready".to_vec(),
                    },
                });
            }
            let response = ReadStreamResponse {
                placement,
                offset: request.offset,
                next_offset: request.offset,
                content_type: DEFAULT_CONTENT_TYPE.to_owned(),
                payload: Vec::new(),
                up_to_date: true,
                closed: false,
                retained_record_range: None,
                record_range: None,
            };
            Ok(GroupReadStreamParts::from_response(response))
        })
    }

    fn touch_stream_access<'a>(
        &'a mut self,
        stream_id: BucketStreamId,
        now_ms: u64,
        renew_ttl: bool,
        placement: ShardPlacement,
    ) -> GroupTouchStreamAccessFuture<'a> {
        self.inner
            .touch_stream_access(stream_id, now_ms, renew_ttl, placement)
    }

    fn close_stream<'a>(
        &'a mut self,
        request: CloseStreamRequest,
        placement: ShardPlacement,
    ) -> GroupCloseStreamFuture<'a> {
        self.inner.close_stream(request, placement)
    }

    fn delete_stream<'a>(
        &'a mut self,
        request: DeleteStreamRequest,
        placement: ShardPlacement,
    ) -> GroupDeleteStreamFuture<'a> {
        self.inner.delete_stream(request, placement)
    }

    fn append<'a>(
        &'a mut self,
        request: AppendRequest,
        placement: ShardPlacement,
        admission: ColdWriteAdmission,
    ) -> GroupAppendFuture<'a> {
        self.inner.append(request, placement, admission)
    }

    fn append_batch<'a>(
        &'a mut self,
        request: AppendBatchRequest,
        placement: ShardPlacement,
        admission: ColdWriteAdmission,
    ) -> GroupAppendBatchFuture<'a> {
        self.inner.append_batch(request, placement, admission)
    }

    fn snapshot<'a>(&'a mut self, placement: ShardPlacement) -> GroupSnapshotFuture<'a> {
        Box::pin(async move {
            Ok(GroupSnapshot {
                placement,
                group_commit_index: 0,
                stream_snapshot: StreamSnapshot {
                    buckets: Vec::new(),
                    erased_buckets: Vec::new(),
                    streams: Vec::new(),
                    pending_cold_gc: Vec::new(),
                    next_cold_gc_seq: 0,
                    shared_cold_object_owners: Vec::new(),
                    bucket_usage: Vec::new(),
                    bucket_quotas: Vec::new(),
                },
                stream_append_counts: Vec::new(),
            })
        })
    }

    fn install_snapshot<'a>(
        &'a mut self,
        _snapshot: GroupSnapshot,
    ) -> GroupInstallSnapshotFuture<'a> {
        Box::pin(async { Ok(()) })
    }
}

impl GroupEngine for RecordingEngine {
    fn bucket_usage<'a>(&'a mut self, _placement: ShardPlacement) -> GroupBucketUsageFuture<'a> {
        Box::pin(async { Ok(Vec::new()) })
    }

    fn accepts_local_writes(&self) -> bool {
        self.accepts_local_writes
    }

    fn create_stream<'a>(
        &'a mut self,
        request: CreateStreamRequest,
        placement: ShardPlacement,
        _admission: ColdWriteAdmission,
    ) -> GroupCreateStreamFuture<'a> {
        Box::pin(async move {
            assert_eq!(placement, self.placement);
            self.commit_index += 1;
            Ok(CreateStreamResponse {
                placement,
                next_offset: u64::try_from(request.initial_payload.len())
                    .expect("payload len fits u64"),
                closed: request.close_after,
                already_exists: false,
                group_commit_index: self.commit_index,
                record_range: None,
            })
        })
    }

    fn head_stream<'a>(
        &'a mut self,
        request: HeadStreamRequest,
        placement: ShardPlacement,
    ) -> GroupHeadStreamFuture<'a> {
        Box::pin(async move {
            assert_eq!(placement, self.placement);
            Ok(HeadStreamResponse {
                placement,
                content_type: DEFAULT_CONTENT_TYPE.to_owned(),
                tail_offset: request.stream_id.stream_id.len() as u64,
                cold_hot_start_offset: 0,
                closed: false,
                stream_ttl_seconds: None,
                stream_expires_at_ms: None,
                snapshot_offset: None,
                snapshot_digest: None,
                retained_offset: 0,
                integrity: empty_integrity(),
                record_range: None,
            })
        })
    }

    fn read_stream<'a>(
        &'a mut self,
        request: ReadStreamRequest,
        placement: ShardPlacement,
    ) -> GroupReadStreamFuture<'a> {
        Box::pin(async move {
            assert_eq!(placement, self.placement);
            Ok(ReadStreamResponse {
                placement,
                offset: request.offset,
                next_offset: request.offset,
                content_type: DEFAULT_CONTENT_TYPE.to_owned(),
                payload: Vec::new(),
                up_to_date: true,
                closed: false,
                retained_record_range: None,
                record_range: None,
            })
        })
    }

    fn touch_stream_access<'a>(
        &'a mut self,
        _stream_id: BucketStreamId,
        _now_ms: u64,
        _renew_ttl: bool,
        placement: ShardPlacement,
    ) -> GroupTouchStreamAccessFuture<'a> {
        Box::pin(async move {
            assert_eq!(placement, self.placement);
            Ok(TouchStreamAccessResponse {
                placement,
                changed: false,
                expired: false,
                group_commit_index: self.commit_index,
            })
        })
    }

    fn close_stream<'a>(
        &'a mut self,
        _request: CloseStreamRequest,
        placement: ShardPlacement,
    ) -> GroupCloseStreamFuture<'a> {
        Box::pin(async move {
            assert_eq!(placement, self.placement);
            self.commit_index += 1;
            Ok(CloseStreamResponse {
                placement,
                next_offset: self.commit_index,
                group_commit_index: self.commit_index,
                deduplicated: false,
                record_range: None,
            })
        })
    }

    fn delete_stream<'a>(
        &'a mut self,
        _request: DeleteStreamRequest,
        placement: ShardPlacement,
    ) -> GroupDeleteStreamFuture<'a> {
        Box::pin(async move {
            assert_eq!(placement, self.placement);
            self.commit_index += 1;
            Ok(DeleteStreamResponse {
                placement,
                group_commit_index: self.commit_index,
            })
        })
    }

    fn append<'a>(
        &'a mut self,
        request: AppendRequest,
        placement: ShardPlacement,
        _admission: ColdWriteAdmission,
    ) -> GroupAppendFuture<'a> {
        Box::pin(async move {
            assert_eq!(placement, self.placement);
            let start_offset = self.commit_index;
            let next_offset = start_offset + request.payload_len();
            self.commit_index += 1;
            Ok(AppendResponse {
                placement,
                start_offset,
                next_offset,
                stream_append_count: self.commit_index,
                group_commit_index: self.commit_index,
                closed: request.close_after,
                deduplicated: false,
                producer: request.producer,
                record_range: None,
                stream_hot_bytes: 0,
                group_hot_bytes: 0,
            })
        })
    }

    fn append_batch<'a>(
        &'a mut self,
        request: AppendBatchRequest,
        placement: ShardPlacement,
        _admission: ColdWriteAdmission,
    ) -> GroupAppendBatchFuture<'a> {
        Box::pin(async move {
            assert_eq!(placement, self.placement);
            let AppendBatchRequest {
                stream_id: _,
                content_type: _,
                payloads,
                producer: _,
                now_ms: _,
            } = request;
            let mut items = Vec::with_capacity(payloads.len());
            for payload in payloads {
                let start_offset = self.commit_index;
                let next_offset =
                    start_offset + u64::try_from(payload.len()).expect("payload len fits u64");
                self.commit_index += 1;
                items.push(Ok(AppendResponse {
                    placement,
                    start_offset,
                    next_offset,
                    stream_append_count: self.commit_index,
                    group_commit_index: self.commit_index,
                    closed: false,
                    deduplicated: false,
                    producer: None,
                    record_range: None,
                    stream_hot_bytes: 0,
                    group_hot_bytes: 0,
                }));
            }
            Ok(GroupAppendBatchResponse { placement, items })
        })
    }

    fn cold_hot_backlog<'a>(
        &'a mut self,
        stream_id: BucketStreamId,
        placement: ShardPlacement,
    ) -> GroupColdHotBacklogFuture<'a> {
        Box::pin(async move {
            assert_eq!(placement, self.placement);
            Ok(ColdHotBacklog {
                stream_id,
                stream_hot_bytes: self.cold_hot_bytes,
                group_hot_bytes: self.cold_hot_bytes,
            })
        })
    }

    fn snapshot<'a>(&'a mut self, placement: ShardPlacement) -> GroupSnapshotFuture<'a> {
        Box::pin(async move {
            assert_eq!(placement, self.placement);
            Ok(GroupSnapshot {
                placement,
                group_commit_index: self.commit_index,
                stream_snapshot: StreamSnapshot {
                    buckets: Vec::new(),
                    erased_buckets: Vec::new(),
                    streams: Vec::new(),
                    pending_cold_gc: Vec::new(),
                    next_cold_gc_seq: 0,
                    shared_cold_object_owners: Vec::new(),
                    bucket_usage: Vec::new(),
                    bucket_quotas: Vec::new(),
                },
                stream_append_counts: Vec::new(),
            })
        })
    }

    fn install_snapshot<'a>(
        &'a mut self,
        snapshot: GroupSnapshot,
    ) -> GroupInstallSnapshotFuture<'a> {
        Box::pin(async move {
            assert_eq!(snapshot.placement, self.placement);
            self.commit_index = snapshot.group_commit_index;
            Ok(())
        })
    }
}

#[derive(Debug, Clone)]
struct BlockingFirstCreateEngineFactory {
    first_create_blocks: Arc<AtomicBool>,
    entered: Arc<Notify>,
    release: Arc<Notify>,
}

impl Default for BlockingFirstCreateEngineFactory {
    fn default() -> Self {
        Self {
            first_create_blocks: Arc::new(AtomicBool::new(true)),
            entered: Arc::new(Notify::new()),
            release: Arc::new(Notify::new()),
        }
    }
}

impl GroupEngineFactory for BlockingFirstCreateEngineFactory {
    fn create<'a>(
        &'a self,
        _placement: ShardPlacement,
        _metrics: GroupEngineMetrics,
    ) -> GroupEngineCreateFuture<'a> {
        Box::pin(async move {
            let engine: Box<dyn GroupEngine> = Box::new(BlockingFirstCreateEngine {
                inner: InMemoryGroupEngine::default(),
                first_create_blocks: self.first_create_blocks.clone(),
                entered: self.entered.clone(),
                release: self.release.clone(),
            });
            Ok(engine)
        })
    }
}

struct BlockingFirstCreateEngine {
    inner: InMemoryGroupEngine,
    first_create_blocks: Arc<AtomicBool>,
    entered: Arc<Notify>,
    release: Arc<Notify>,
}

impl GroupEngine for BlockingFirstCreateEngine {
    fn create_stream<'a>(
        &'a mut self,
        request: CreateStreamRequest,
        placement: ShardPlacement,
        admission: ColdWriteAdmission,
    ) -> GroupCreateStreamFuture<'a> {
        let should_block = self.first_create_blocks.swap(false, Ordering::SeqCst);
        let entered = self.entered.clone();
        let release = self.release.clone();
        Box::pin(async move {
            if should_block {
                entered.notify_one();
                release.notified().await;
            }
            self.inner
                .create_stream(request, placement, admission)
                .await
        })
    }

    fn head_stream<'a>(
        &'a mut self,
        request: HeadStreamRequest,
        placement: ShardPlacement,
    ) -> GroupHeadStreamFuture<'a> {
        self.inner.head_stream(request, placement)
    }

    fn bucket_usage<'a>(&'a mut self, placement: ShardPlacement) -> GroupBucketUsageFuture<'a> {
        self.inner.bucket_usage(placement)
    }

    fn read_stream<'a>(
        &'a mut self,
        request: ReadStreamRequest,
        placement: ShardPlacement,
    ) -> GroupReadStreamFuture<'a> {
        self.inner.read_stream(request, placement)
    }

    fn touch_stream_access<'a>(
        &'a mut self,
        stream_id: BucketStreamId,
        now_ms: u64,
        renew_ttl: bool,
        placement: ShardPlacement,
    ) -> GroupTouchStreamAccessFuture<'a> {
        self.inner
            .touch_stream_access(stream_id, now_ms, renew_ttl, placement)
    }

    fn close_stream<'a>(
        &'a mut self,
        request: CloseStreamRequest,
        placement: ShardPlacement,
    ) -> GroupCloseStreamFuture<'a> {
        self.inner.close_stream(request, placement)
    }

    fn delete_stream<'a>(
        &'a mut self,
        request: DeleteStreamRequest,
        placement: ShardPlacement,
    ) -> GroupDeleteStreamFuture<'a> {
        self.inner.delete_stream(request, placement)
    }

    fn append<'a>(
        &'a mut self,
        request: AppendRequest,
        placement: ShardPlacement,
        admission: ColdWriteAdmission,
    ) -> GroupAppendFuture<'a> {
        self.inner.append(request, placement, admission)
    }

    fn append_batch<'a>(
        &'a mut self,
        request: AppendBatchRequest,
        placement: ShardPlacement,
        admission: ColdWriteAdmission,
    ) -> GroupAppendBatchFuture<'a> {
        self.inner.append_batch(request, placement, admission)
    }

    fn snapshot<'a>(&'a mut self, placement: ShardPlacement) -> GroupSnapshotFuture<'a> {
        self.inner.snapshot(placement)
    }

    fn install_snapshot<'a>(
        &'a mut self,
        snapshot: GroupSnapshot,
    ) -> GroupInstallSnapshotFuture<'a> {
        self.inner.install_snapshot(snapshot)
    }
}

#[derive(Debug, Clone)]
struct BlockingOnceFactory {
    first_create_blocks: Arc<AtomicBool>,
    entered: Arc<Notify>,
    release: Arc<Notify>,
}

impl Default for BlockingOnceFactory {
    fn default() -> Self {
        Self {
            first_create_blocks: Arc::new(AtomicBool::new(true)),
            entered: Arc::new(Notify::new()),
            release: Arc::new(Notify::new()),
        }
    }
}

impl GroupEngineFactory for BlockingOnceFactory {
    fn create<'a>(
        &'a self,
        _placement: ShardPlacement,
        _metrics: GroupEngineMetrics,
    ) -> GroupEngineCreateFuture<'a> {
        Box::pin(async move {
            if self.first_create_blocks.swap(false, Ordering::SeqCst) {
                self.entered.notify_one();
                self.release.notified().await;
            }
            let engine: Box<dyn GroupEngine> = Box::new(InMemoryGroupEngine::default());
            Ok(engine)
        })
    }
}

#[derive(Debug, Clone, Copy)]
struct FailingFactory;

impl GroupEngineFactory for FailingFactory {
    fn create<'a>(
        &'a self,
        _placement: ShardPlacement,
        _metrics: GroupEngineMetrics,
    ) -> GroupEngineCreateFuture<'a> {
        Box::pin(async {
            let engine: Box<dyn GroupEngine> = Box::new(FailingEngine);
            Ok(engine)
        })
    }
}

struct FailingEngine;

impl GroupEngine for FailingEngine {
    fn bucket_usage<'a>(&'a mut self, _placement: ShardPlacement) -> GroupBucketUsageFuture<'a> {
        Box::pin(async { Ok(Vec::new()) })
    }

    fn create_stream<'a>(
        &'a mut self,
        _request: CreateStreamRequest,
        _placement: ShardPlacement,
        _admission: ColdWriteAdmission,
    ) -> GroupCreateStreamFuture<'a> {
        Box::pin(async { Err(GroupEngineError::new("proposal rejected")) })
    }

    fn head_stream<'a>(
        &'a mut self,
        _request: HeadStreamRequest,
        _placement: ShardPlacement,
    ) -> GroupHeadStreamFuture<'a> {
        Box::pin(async { Err(GroupEngineError::new("proposal rejected")) })
    }

    fn read_stream<'a>(
        &'a mut self,
        _request: ReadStreamRequest,
        _placement: ShardPlacement,
    ) -> GroupReadStreamFuture<'a> {
        Box::pin(async { Err(GroupEngineError::new("proposal rejected")) })
    }

    fn touch_stream_access<'a>(
        &'a mut self,
        _stream_id: BucketStreamId,
        _now_ms: u64,
        _renew_ttl: bool,
        _placement: ShardPlacement,
    ) -> GroupTouchStreamAccessFuture<'a> {
        Box::pin(async { Err(GroupEngineError::new("proposal rejected")) })
    }

    fn close_stream<'a>(
        &'a mut self,
        _request: CloseStreamRequest,
        _placement: ShardPlacement,
    ) -> GroupCloseStreamFuture<'a> {
        Box::pin(async { Err(GroupEngineError::new("proposal rejected")) })
    }

    fn delete_stream<'a>(
        &'a mut self,
        _request: DeleteStreamRequest,
        _placement: ShardPlacement,
    ) -> GroupDeleteStreamFuture<'a> {
        Box::pin(async { Err(GroupEngineError::new("proposal rejected")) })
    }

    fn append<'a>(
        &'a mut self,
        _request: AppendRequest,
        _placement: ShardPlacement,
        _admission: ColdWriteAdmission,
    ) -> GroupAppendFuture<'a> {
        Box::pin(async { Err(GroupEngineError::new("proposal rejected")) })
    }

    fn append_batch<'a>(
        &'a mut self,
        _request: AppendBatchRequest,
        _placement: ShardPlacement,
        _admission: ColdWriteAdmission,
    ) -> GroupAppendBatchFuture<'a> {
        Box::pin(async { Err(GroupEngineError::new("proposal rejected")) })
    }

    fn snapshot<'a>(&'a mut self, _placement: ShardPlacement) -> GroupSnapshotFuture<'a> {
        Box::pin(async { Err(GroupEngineError::new("proposal rejected")) })
    }

    fn install_snapshot<'a>(
        &'a mut self,
        _snapshot: GroupSnapshot,
    ) -> GroupInstallSnapshotFuture<'a> {
        Box::pin(async { Err(GroupEngineError::new("proposal rejected")) })
    }
}