automerge 0.10.0

//! # Sync Protocol
//!
//! The sync protocol is based on this paper:
//! <https://arxiv.org/abs/2012.00472>, it assumes a reliable in-order stream
//! between two peers who are synchronizing a document.
//!
//! Each peer maintains a [`State`] for each peer they are synchronizing with.
//! This state tracks things like what the heads of the other peer are and
//! whether there are in-flight messages. Anything which implements [`SyncDoc`]
//! can take part in the sync protocol. The flow goes something like this:
//!
//! * The initiating peer creates an empty [`State`] and then calls
//!   [`SyncDoc::generate_sync_message()`] to generate new sync message and sends
//!   it to the receiving peer.
//! * The receiving peer receives a message from the initiator, creates a new
//!   [`State`], and calls [`SyncDoc::receive_sync_message()`] on it's view of the
//!   document
//! * The receiving peer then calls [`SyncDoc::generate_sync_message()`] to generate
//!   a new sync message and send it back to the initiator
//! * From this point on each peer operates in a loop, receiving a sync message
//!   from the other peer and then generating a new message to send back.
//!
//! ## Example
//!
//! ```
//! use automerge::{transaction::Transactable, sync::{self, SyncDoc}, ReadDoc};
//! # fn main() -> Result<(), automerge::AutomergeError> {
//! // Create a document on peer1
//! let mut peer1 = automerge::AutoCommit::new();
//! peer1.put(automerge::ROOT, "key", "value")?;
//!
//! // Create a state to track our sync with peer2
//! let mut peer1_state = sync::State::new();
//! // Generate the initial message to send to peer2, unwrap for brevity
//! let message1to2 = peer1.sync().generate_sync_message(&mut peer1_state).unwrap();
//!
//! // We receive the message on peer2. We don't have a document at all yet
//! // so we create one
//! let mut peer2 = automerge::AutoCommit::new();
//! // We don't have a state for peer1 (it's a new connection), so we create one
//! let mut peer2_state = sync::State::new();
//! // Now receive the message from peer 1
//! peer2.sync().receive_sync_message(&mut peer2_state, message1to2)?;
//!
//! // Now we loop, sending messages from one to two and two to one until
//! // neither has anything new to send
//!
//! loop {
//!     let two_to_one = peer2.sync().generate_sync_message(&mut peer2_state);
//!     if let Some(message) = two_to_one.as_ref() {
//!         println!("two to one");
//!         peer1.sync().receive_sync_message(&mut peer1_state, message.clone())?;
//!     }
//!     let one_to_two = peer1.sync().generate_sync_message(&mut peer1_state);
//!     if let Some(message) = one_to_two.as_ref() {
//!         println!("one to two");
//!         peer2.sync().receive_sync_message(&mut peer2_state, message.clone())?;
//!     }
//!     if two_to_one.is_none() && one_to_two.is_none() {
//!         break;
//!     }
//! }
//!
//! assert_eq!(peer2.get(automerge::ROOT, "key")?.unwrap().0.to_str(), Some("value"));
//!
//! # Ok(())
//! # }
//! ```

use itertools::Itertools;
use serde::ser::SerializeMap;
use std::collections::{HashMap, HashSet};

use crate::{
    patches::PatchLog,
    storage::{parse, ReadChangeOpError},
    Automerge, AutomergeError, ChangeHash, ReadDoc,
};

mod bloom;
mod message_builder;
mod state;
use message_builder::MessageBuilder;

#[cfg(test)]
mod v1_compat_test;

pub use bloom::{BloomFilter, DecodeError as DecodeBloomError};
pub use state::DecodeError as DecodeStateError;
pub use state::{Have, State};

/// A document which can take part in the sync protocol
///
/// See the [module level documentation](crate::sync) for more details.
pub trait SyncDoc {
    /// Generate a sync message for the remote peer represented by `sync_state`
    ///
    /// If this returns [`None`] then there are no new messages to send, either because we are
    /// waiting for an acknolwedgement of an in-flight message, or because the remote is up to
    /// date.
    ///
    /// * `sync_state` - The [`State`] for this document and the remote peer
    /// * `message` - The [`Message`] to receive
    /// * `patch_log` - A [`PatchLog`] which will be updated with any changes that are made to the current state of the document due to the received sync message
    fn generate_sync_message(&self, sync_state: &mut State) -> Option<Message>;

    /// Apply a received sync message to this document and `sync_state`
    fn receive_sync_message(
        &mut self,
        sync_state: &mut State,
        message: Message,
    ) -> Result<(), AutomergeError>;

    /// Apply a received sync message to this document and `sync_state`, logging any changes that
    /// are made to `patch_log`
    ///
    /// If this returns [`None`] then there are no new messages to send, either because we are
    /// waiting for an acknolwedgement of an in-flight message, or because the remote is up to
    /// date.
    ///
    /// # Arguments
    ///
    /// * `sync_state` - The [`State`] for this document and the remote peer
    /// * `message` - The [`Message`] to receive
    /// * `patch_log` - A [`PatchLog`] which will be updated with any changes that are made to the current state of the document due to the received sync message
    fn receive_sync_message_log_patches(
        &mut self,
        sync_state: &mut State,
        message: Message,
        patch_log: &mut PatchLog,
    ) -> Result<(), AutomergeError>;
}

const MESSAGE_TYPE_SYNC: u8 = 0x42; // first byte of a sync message, for identification
const MESSAGE_TYPE_SYNC_V2: u8 = 0x43; // first byte of a sync message, for identification

#[derive(Clone, Debug, PartialEq)]
pub enum MessageVersion {
    V1,
    V2,
}

impl MessageVersion {
    fn parse(input: parse::Input<'_>) -> parse::ParseResult<'_, Self, ReadMessageError> {
        let (i, first_byte) = parse::take1(input)?;
        match first_byte {
            MESSAGE_TYPE_SYNC => Ok((i, Self::V1)),
            MESSAGE_TYPE_SYNC_V2 => Ok((i, Self::V2)),
            _ => Err(parse::ParseError::Error(ReadMessageError::WrongType {
                expected_one_of: vec![MESSAGE_TYPE_SYNC, MESSAGE_TYPE_SYNC_V2],
                found: first_byte,
            })),
        }
    }

    fn encode(&self) -> u8 {
        match self {
            Self::V1 => MESSAGE_TYPE_SYNC,
            Self::V2 => MESSAGE_TYPE_SYNC_V2,
        }
    }
}

impl SyncDoc for Automerge {
    fn generate_sync_message(&self, sync_state: &mut State) -> Option<Message> {
        let our_heads = self.get_heads();

        let our_need = if sync_state.read_only {
            vec![]
        } else {
            self.get_missing_deps(sync_state.their_heads.as_ref().unwrap_or(&vec![]))
        };

        let their_heads_set = if let Some(ref heads) = sync_state.their_heads {
            heads.iter().collect::<HashSet<_>>()
        } else {
            HashSet::new()
        };
        let our_have = if our_need.iter().all(|hash| their_heads_set.contains(hash)) {
            vec![self.make_bloom_filter(sync_state.shared_heads.clone())]
        } else {
            Vec::new()
        };

        if let Some(ref their_have) = sync_state.their_have {
            if let Some(first_have) = their_have.first().as_ref() {
                if !first_have
                    .last_sync
                    .iter()
                    .all(|hash| self.has_change(hash))
                {
                    return Some(Message::reset(our_heads));
                }
            }
        }

        let message_builder = if sync_state.is_peer_read_only() {
            // The remote peer is read-only and will ignore incoming changes.
            // Skip computing and sending changes to save bandwidth.
            MessageBuilder::new(vec![], sync_state)
        } else if let Some((their_have, their_need)) = sync_state.their() {
            if sync_state.send_doc() {
                let hashes = self.change_graph.get_hashes(&[]);
                MessageBuilder::new_v2(self.save(), hashes)
            } else {
                let all_hashes = self
                    .get_hashes_to_send(their_have, their_need)
                    .expect("Should have only used hashes that are in the document");
                // deduplicate the changes to send with those we have already sent and clone it now
                let hashes: Vec<_> = all_hashes
                    .into_iter()
                    .filter(|hash| !sync_state.sent_hashes.contains(hash))
                    .collect();
                if hashes.len() > self.change_graph.len() / 3 && sync_state.supports_v2_messages() {
                    // sending more than a 1/3 of the document?  send everything
                    let all_hashes = self.change_graph.get_hashes(&[]);
                    MessageBuilder::new_v2(self.save(), all_hashes)
                } else {
                    let changes = self.get_changes_by_hashes(hashes.iter().copied()).ok()?;
                    MessageBuilder::new(changes, sync_state)
                }
            }
        } else {
            MessageBuilder::new(vec![], sync_state)
        };

        let heads_unchanged = sync_state.last_sent_heads == our_heads;

        let heads_equal = sync_state.their_heads.as_ref() == Some(&our_heads);

        if heads_unchanged && sync_state.have_responded {
            if (heads_equal || sync_state.read_only) && message_builder.is_empty() {
                return None;
            }
            if sync_state.in_flight {
                return None;
            }
        }

        sync_state.have_responded = true;
        sync_state.last_sent_heads.clone_from(&our_heads);
        sync_state.sent_hashes.extend(message_builder.hashes());

        let mut flags = MessageFlags::new();
        flags.set(MessageFlags::SUPPORTS_SYNC_RESET);
        if sync_state.read_only {
            flags.set(MessageFlags::READ_ONLY);
        }

        // When switching from read-only to read-write, we need the remote to
        // clear its sent_hashes so it resends changes we previously ignored.
        // Peers that advertise Capability::SyncReset understand the SyncReset
        // flag. Old peers don't, so we send empty heads to simulate losing all
        // local state, which triggers the same sent_hashes clearing via the
        // existing "peer lost all data" code path.
        let heads_to_send = if sync_state.needs_reset {
            sync_state.needs_reset = false;
            if sync_state.peer_supports_sync_reset() {
                flags.set(MessageFlags::SYNC_RESET);
                our_heads.clone()
            } else {
                vec![]
            }
        } else {
            our_heads.clone()
        };

        let sync_message = message_builder
            .heads(heads_to_send)
            .have(our_have)
            .need(our_need)
            .flags(Some(flags))
            .build();

        sync_state.in_flight = true;
        Some(sync_message)
    }

    fn receive_sync_message(
        &mut self,
        sync_state: &mut State,
        message: Message,
    ) -> Result<(), AutomergeError> {
        let mut patch_log = PatchLog::inactive();
        self.receive_sync_message_inner(sync_state, message, &mut patch_log)
    }

    fn receive_sync_message_log_patches(
        &mut self,
        sync_state: &mut State,
        message: Message,
        patch_log: &mut PatchLog,
    ) -> Result<(), AutomergeError> {
        self.receive_sync_message_inner(sync_state, message, patch_log)
    }
}

impl Automerge {
    #[inline(never)]
    fn make_bloom_filter(&self, last_sync: Vec<ChangeHash>) -> Have {
        let hashes = self.change_graph.get_hashes(&last_sync);
        Have {
            last_sync,
            bloom: BloomFilter::from_hashes(hashes.iter()),
        }
    }

    #[inline(never)]
    fn get_hashes_to_send(
        &self,
        have: &[Have],
        need: &[ChangeHash],
    ) -> Result<Vec<ChangeHash>, AutomergeError> {
        if have.is_empty() {
            Ok(need.to_vec())
        } else {
            let mut last_sync_hashes = HashSet::new();
            let mut bloom_filters = Vec::with_capacity(have.len());

            for h in have {
                let Have { last_sync, bloom } = h;
                last_sync_hashes.extend(last_sync);
                bloom_filters.push(bloom);
            }
            let last_sync_hashes = last_sync_hashes.into_iter().copied().collect::<Vec<_>>();

            let hashes = self.change_graph.get_hashes(&last_sync_hashes);

            let mut change_hashes = HashSet::with_capacity(hashes.len());
            let mut dependents: HashMap<ChangeHash, Vec<ChangeHash>> = HashMap::new();
            let mut hashes_to_send = HashSet::new();

            for hash in &*hashes {
                change_hashes.insert(*hash);

                for dep in self.change_graph.deps(hash) {
                    dependents.entry(dep).or_default().push(*hash);
                }

                if bloom_filters.iter().all(|bloom| !bloom.contains_hash(hash)) {
                    hashes_to_send.insert(*hash);
                }
            }

            let mut stack = hashes_to_send.iter().copied().collect::<Vec<_>>();
            while let Some(hash) = stack.pop() {
                if let Some(deps) = dependents.get(&hash) {
                    for dep in deps {
                        if hashes_to_send.insert(*dep) {
                            stack.push(*dep);
                        }
                    }
                }
            }

            let mut final_hashes = Vec::with_capacity(hashes_to_send.len() + need.len());
            for hash in need {
                if !hashes_to_send.contains(hash) {
                    final_hashes.push(*hash);
                }
            }

            for hash in &*hashes {
                if hashes_to_send.contains(hash) {
                    final_hashes.push(*hash);
                }
            }
            Ok(final_hashes)
        }
    }

    #[inline(never)]
    pub(crate) fn receive_sync_message_inner(
        &mut self,
        sync_state: &mut State,
        message: Message,
        patch_log: &mut PatchLog,
    ) -> Result<(), AutomergeError> {
        sync_state.in_flight = false;
        let before_heads = self.get_heads();

        let Message {
            heads: message_heads,
            changes: message_changes,
            need: message_need,
            have: message_have,
            flags: message_flags,
            ..
        } = message;

        if let Some(flags) = message_flags {
            // Any peer that sends the flags section supports V2 messages —
            // the flags section was introduced alongside V2 support.
            let mut caps = vec![Capability::MessageV2];
            if flags.contains(MessageFlags::SUPPORTS_SYNC_RESET) {
                caps.push(Capability::SyncReset);
            }
            sync_state.their_capabilities = Some(caps);

            // Process transient per-message signals
            if flags.contains(MessageFlags::SYNC_RESET) {
                sync_state.sent_hashes.clear();
            }
            sync_state.peer_read_only = flags.contains(MessageFlags::READ_ONLY);
        }

        let changes_is_empty = message_changes.is_empty();
        if !changes_is_empty && !sync_state.read_only {
            self.load_incremental_log_patches(&message_changes.join(), patch_log)?;
            sync_state.shared_heads = advance_heads(
                &before_heads.iter().collect(),
                &self.get_heads().into_iter().collect(),
                &sync_state.shared_heads,
            );
        }

        // trim down the sent hashes to those that we know they haven't seen
        self.filter_changes(&message_heads, &mut sync_state.sent_hashes)?;

        if changes_is_empty && message_heads == before_heads {
            sync_state.last_sent_heads.clone_from(&message_heads);
        }

        let known_heads = message_heads
            .iter()
            .filter(|head| self.has_change(head))
            .collect::<Vec<_>>();
        if known_heads.len() == message_heads.len() {
            sync_state.shared_heads.clone_from(&message_heads);
            // If the remote peer has lost all its data, reset our state to perform a full resync
            if message_heads.is_empty() {
                sync_state.last_sent_heads = Default::default();
                sync_state.sent_hashes = Default::default();
            }
        } else {
            sync_state.shared_heads = sync_state
                .shared_heads
                .iter()
                .chain(known_heads)
                .copied()
                .unique()
                .sorted()
                .collect::<Vec<_>>();
        }

        sync_state.their_have = Some(message_have);
        sync_state.their_heads = Some(message_heads);
        sync_state.their_need = Some(message_need);

        Ok(())
    }
}

#[derive(Debug, thiserror::Error)]
pub enum ReadMessageError {
    #[error("expected {expected_one_of:?} but found {found}")]
    WrongType { expected_one_of: Vec<u8>, found: u8 },
    #[error("{0}")]
    Parse(String),
    #[error(transparent)]
    ReadChangeOps(#[from] ReadChangeOpError),
    #[error("not enough input")]
    NotEnoughInput,
}

impl From<parse::leb128::Error> for ReadMessageError {
    fn from(e: parse::leb128::Error) -> Self {
        ReadMessageError::Parse(e.to_string())
    }
}

impl From<bloom::ParseError> for ReadMessageError {
    fn from(e: bloom::ParseError) -> Self {
        ReadMessageError::Parse(e.to_string())
    }
}

impl From<crate::storage::change::ParseError> for ReadMessageError {
    fn from(e: crate::storage::change::ParseError) -> Self {
        ReadMessageError::Parse(format!("error parsing changes: {}", e))
    }
}

impl From<ReadMessageError> for parse::ParseError<ReadMessageError> {
    fn from(e: ReadMessageError) -> Self {
        parse::ParseError::Error(e)
    }
}

impl From<parse::ParseError<ReadMessageError>> for ReadMessageError {
    fn from(p: parse::ParseError<ReadMessageError>) -> Self {
        match p {
            parse::ParseError::Error(e) => e,
            parse::ParseError::Incomplete(..) => Self::NotEnoughInput,
        }
    }
}

/// The sync message to be sent.
///
/// ## Notes about encoding
///
/// There are two versions of the sync message, V1 and V2. The V1 message is the original message
/// which automerge shipped with and V2 is an extension which allows for encoding the changes as
/// either a list of changes or as a compressed document format. This makes syncing up for the
/// first time faster.
///
/// Encoding this in a backwards compatible way is a bit tricky. The wire format of the v1 message
/// didn't allow for any forwards compatible changes. In order to accomodate this the first message
/// a peer sends is a V1 message with a length-prefixed `Vec<MessageFlag>` appended to it. For old
/// implementations this appended data is just ignored but new implementations read it and store
/// the advertised capabilities on the sync state. This allows new implementations to discover if
/// the remote peer supports the V2 message format and if so send a V2 message. The flags also
/// carry transient per-message signals such as read-only mode and sync reset.
#[derive(Clone, Debug, PartialEq)]
pub struct Message {
    /// The heads of the sender.
    pub heads: Vec<ChangeHash>,
    /// The hashes of any changes that are being explicitly requested from the recipient.
    pub need: Vec<ChangeHash>,
    /// A summary of the changes that the sender already has.
    pub have: Vec<Have>,
    /// The changes for the recipient to apply.
    ///
    /// This is a Vec of bytes which should be passed to `Automerge::load_incremental`. The reason
    /// it is a `Vec<Vec<u8>>` and not a `Vec<u8>` is that the V1 message format is a sequence of
    /// change chunks, each of which is length delimited. The V2 message format is a single length
    /// delimited chunk but we nest it inside a Vec for backwards compatibility.
    pub changes: ChunkList,
    /// Per-message flags including capability advertisements and transient signals.
    pub flags: Option<MessageFlags>,
    /// What version to encode this message as
    pub version: MessageVersion,
}

/// An array of changes, each of which should be passed to [`Automerge::load_incremental()`]
#[derive(Clone, Debug, PartialEq)]
pub struct ChunkList(Vec<Vec<u8>>);

impl From<Vec<Vec<u8>>> for ChunkList {
    fn from(v: Vec<Vec<u8>>) -> Self {
        Self(v)
    }
}

impl From<Vec<u8>> for ChunkList {
    fn from(v: Vec<u8>) -> Self {
        Self(vec![v])
    }
}

impl ChunkList {
    fn parse(i: parse::Input<'_>) -> parse::ParseResult<'_, Self, ReadMessageError> {
        let change_parser = |i| {
            let (i, bytes) = parse::length_prefixed_bytes(i)?;
            Ok((i, bytes.to_vec()))
        };
        let (i, stored_changes) = parse::length_prefixed(change_parser)(i)?;
        Ok((i, Self(stored_changes)))
    }

    pub fn empty() -> Self {
        Self(Vec::new())
    }

    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }

    pub fn len(&self) -> usize {
        self.0.len()
    }

    pub fn iter(&self) -> impl ExactSizeIterator<Item = &[u8]> {
        self.0.iter().map(|v| v.as_slice())
    }

    pub(crate) fn join(&self) -> Vec<u8> {
        let total: usize = self.0.iter().map(Vec::len).sum();
        let mut result = Vec::with_capacity(total);

        for v in &self.0 {
            result.extend(v);
        }

        result
    }
}

impl serde::Serialize for Message {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        let mut map = serializer.serialize_map(Some(4))?;
        map.serialize_entry("heads", &self.heads)?;
        map.serialize_entry("need", &self.need)?;
        map.serialize_entry("have", &self.have)?;
        map.serialize_entry("changes", &self.changes.0)?;
        map.end()
    }
}

fn parse_have(input: parse::Input<'_>) -> parse::ParseResult<'_, Have, ReadMessageError> {
    let (i, last_sync) = parse::length_prefixed(parse::change_hash)(input)?;
    let (i, bloom_bytes) = parse::length_prefixed_bytes(i)?;
    let (_, bloom) = BloomFilter::parse(parse::Input::new(bloom_bytes)).map_err(|e| e.lift())?;
    Ok((i, Have { last_sync, bloom }))
}

impl Message {
    pub(crate) fn reset(our_heads: Vec<ChangeHash>) -> Message {
        Message {
            heads: our_heads,
            need: Vec::new(),
            have: vec![Have::default()],
            changes: ChunkList::empty(),
            flags: {
                let mut f = MessageFlags::new();
                f.set(MessageFlags::SUPPORTS_SYNC_RESET);
                Some(f)
            },
            version: MessageVersion::V1,
        }
    }

    pub fn decode(input: &[u8]) -> Result<Self, ReadMessageError> {
        let input = parse::Input::new(input);
        match Self::parse(input) {
            Ok((_, msg)) => Ok(msg),
            Err(parse::ParseError::Error(e)) => Err(e),
            Err(parse::ParseError::Incomplete(_)) => Err(ReadMessageError::NotEnoughInput),
        }
    }

    pub(crate) fn parse(input: parse::Input<'_>) -> parse::ParseResult<'_, Self, ReadMessageError> {
        let (i, message_version) = MessageVersion::parse(input)?;

        let (i, heads) = parse::length_prefixed(parse::change_hash)(i)?;
        let (i, need) = parse::length_prefixed(parse::change_hash)(i)?;
        let (i, have) = parse::length_prefixed(parse_have)(i)?;

        let (i, changes) = ChunkList::parse(i)?;
        let (i, flags) = if !i.is_empty() {
            let (i, raw_bytes) = parse::length_prefixed_bytes(i)?;
            (i, Some(MessageFlags::parse_bytes(raw_bytes)))
        } else {
            (i, None)
        };
        Ok((
            i,
            Message {
                heads,
                need,
                have,
                changes,
                flags,
                version: message_version,
            },
        ))
    }

    pub fn encode(self) -> Vec<u8> {
        let mut buf = vec![self.version.encode()];

        encode_hashes(&mut buf, &self.heads);
        encode_hashes(&mut buf, &self.need);
        encode_many(&mut buf, self.have.iter(), |buf, h| {
            encode_hashes(buf, &h.last_sync);
            leb128::write::unsigned(buf, h.bloom.to_bytes().len() as u64).unwrap();
            buf.extend(h.bloom.to_bytes());
        });

        encode_many(&mut buf, self.changes.iter(), |buf, change| {
            leb128::write::unsigned(buf, change.len() as u64).unwrap();
            buf.extend::<&[u8]>(change.as_ref())
        });

        if let Some(flags) = self.flags {
            flags.encode(&mut buf);
        }

        buf
    }
}

/// Per-message flags packed into a bitfield byte on the wire.
///
/// ## Wire encoding
///
/// Flags are encoded a little strangely, it's easiest to explain by explaining
/// the history. The first flag we added was to indicate that the other end
/// supported "v2" sync protocol messages - sync messages in which the entire
/// document state is sent in a single message. This flag was forward compatibly
/// encoded by taking advantage of the fact that the original parser would ignore
/// any bytes after the end of the sync message. We thus encoded new flags as
/// length prefixed set of flags at the end of the message. I.e. it looked
/// a bit like this:
///
/// | message length | < sync message > | number of flags | flags |
///
/// The idea was that this allows us to forward compatibly add new flags to the
/// protocol by just sending a longer list of flags. The problem with this
/// approach is that it usees a byte per flag, which becomes much more wasteful
/// as the number of flags grows. Thus, we now encode flags as bitfields. We
/// want to stay backwards compatible for implementations unaware of the bitfields
/// though. To do this we take advantage of the fact that there was only ever
/// one flag sent - the old `MessageV2` flag. This means that the only flags
/// ever sent were:
///
/// | 0x01 | 0x02 |
///
/// Where `0x01` is the length prefix byte and `0x02` is the old `MessageV2` flag.
/// To stay backwards compatible we encode the old `MessageV2` flag as a single byte
/// and then all the remaining flags as a byte with the high bit set. The old
/// code thus sees a bunch of unknown bytes after the old MessageV2 flag and
/// ignores them, the new code reads every bit and parses any byte with it's high
/// bit set as a bitfield of flags. This scheme allows us to forward compatibly
/// add as many flags as we need (by increasing the length prefix) without breaking
/// old implementations.
///
/// The final encoding then is
///
/// | length prefix | old MessageV2 flag | bitfield byte (0x80 set) |
///
/// When parsing messages from old implementations (all bytes < 0x80), the
/// old individual byte values are mapped to the corresponding flag bits.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
pub struct MessageFlags(u8);

impl MessageFlags {
    // V1/V2 message format support is not encoded in the bitfield — V2 is
    // communicated via the legacy `0x02` byte that always precedes the
    // bitfield, and V1 is the default. This leaves all 7 bitfield bits
    // available for future flags.

    /// Signals the remote peer should clear its `sent_hashes` and perform a
    /// fresh sync. Used when switching from read-only to read-write mode.
    pub const SYNC_RESET: u8 = 1 << 0;
    /// Indicates the sender is in read-only mode and will not apply incoming
    /// changes.
    pub const READ_ONLY: u8 = 1 << 1;
    /// Advertises that the sender understands the [`SYNC_RESET`](Self::SYNC_RESET)
    /// flag and will clear `sent_hashes` when it receives one.
    pub const SUPPORTS_SYNC_RESET: u8 = 1 << 2;

    const BITFIELD_MARKER: u8 = 0x80;
    /// The old MessageV2 byte, sent first for backwards compatibility.
    const LEGACY_V2_BYTE: u8 = 0x02;

    pub fn new() -> Self {
        Self(0)
    }

    pub fn contains(self, flag: u8) -> bool {
        self.0 & flag != 0
    }

    pub fn set(&mut self, flag: u8) {
        self.0 |= flag;
    }

    fn encode(&self, out: &mut Vec<u8>) {
        // Two entries: 0x02 (old MessageV2 for backwards compat) + bitfield byte
        leb128::write::unsigned(out, 2u64).unwrap();
        out.push(Self::LEGACY_V2_BYTE);
        out.push(Self::BITFIELD_MARKER | self.0);
    }

    fn parse_bytes(bytes: &[u8]) -> Self {
        let mut flags = Self::new();
        for &byte in bytes {
            if byte & Self::BITFIELD_MARKER != 0 {
                // New-style bitfield byte: extract the flag bits
                flags.0 |= byte & !Self::BITFIELD_MARKER;
            }
            // Old-style bytes (0x01, 0x02) are ignored — V1/V2 support
            // is inferred from the presence of the flags section itself
        }
        flags
    }
}

/// Persistent peer capabilities, derived from [`MessageFlags`]s on incoming
/// messages. These describe what the remote peer supports.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum Capability {
    MessageV1,
    MessageV2,
    /// The peer understands the [`MessageFlags::SYNC_RESET`] flag and will
    /// clear its `sent_hashes` when it receives one.
    SyncReset,
}

fn encode_many<'a, I, It, F>(out: &mut Vec<u8>, data: I, f: F)
where
    I: Iterator<Item = It> + ExactSizeIterator + 'a,
    F: Fn(&mut Vec<u8>, It),
{
    leb128::write::unsigned(out, data.len() as u64).unwrap();
    for datum in data {
        f(out, datum)
    }
}

fn encode_hashes(buf: &mut Vec<u8>, hashes: &[ChangeHash]) {
    debug_assert!(
        hashes.windows(2).all(|h| h[0] <= h[1]),
        "hashes were not sorted"
    );
    encode_many(buf, hashes.iter(), |buf, hash| buf.extend(hash.as_bytes()))
}

fn advance_heads(
    my_old_heads: &HashSet<&ChangeHash>,
    my_new_heads: &HashSet<ChangeHash>,
    our_old_shared_heads: &[ChangeHash],
) -> Vec<ChangeHash> {
    let new_heads = my_new_heads
        .iter()
        .filter(|head| !my_old_heads.contains(head))
        .copied()
        .collect::<Vec<_>>();

    let common_heads = our_old_shared_heads
        .iter()
        .filter(|head| my_new_heads.contains(head))
        .copied()
        .collect::<Vec<_>>();

    let mut advanced_heads = HashSet::with_capacity(new_heads.len() + common_heads.len());
    for head in new_heads.into_iter().chain(common_heads) {
        advanced_heads.insert(head);
    }
    let mut advanced_heads = advanced_heads.into_iter().collect::<Vec<_>>();
    advanced_heads.sort();
    advanced_heads
}

#[cfg(test)]
mod tests {
    use super::*;
    //use crate::change::gen::gen_change;
    use crate::storage::parse::Input;
    use crate::storage::Chunk;
    use crate::transaction::Transactable;
    //use crate::types::gen::gen_hash;
    use crate::ActorId;
    //use proptest::prelude::*;

    /*
    prop_compose! {
        fn gen_bloom()(hashes in gen_sorted_hashes(0..10)) -> BloomFilter {
            BloomFilter::from_hashes(hashes.into_iter())
        }
    }

    prop_compose! {
        fn gen_have()(bloom in gen_bloom(), last_sync in gen_sorted_hashes(0..10))  -> Have {
            Have {
                bloom,
                last_sync,
            }
        }
    }

    fn gen_sorted_hashes(size: std::ops::Range<usize>) -> impl Strategy<Value = Vec<ChangeHash>> {
        proptest::collection::vec(gen_hash(), size).prop_map(|mut h| {
            h.sort();
            h
        })
    }

        prop_compose! {
            fn gen_sync_message_v1()(
                heads in gen_sorted_hashes(0..10),
                need in gen_sorted_hashes(0..10),
                have in proptest::collection::vec(gen_have(), 0..10),
                changes in proptest::collection::vec(gen_change(), 0..10),
                supported_capabilities in prop_oneof![
                    Just(None),
                    Just(Some(vec![Capability::MessageV1])),
                    Just(Some(vec![Capability::MessageV2])),
                    Just(Some(vec![Capability::MessageV1, Capability::MessageV2])),
                ],
            ) -> Message {
                Message {
                    heads,
                    need,
                    have,
                    changes: changes.into_iter().map(|c| c.raw_bytes().to_vec()).collect::<Vec<Vec<u8>>>().into(),
                    supported_capabilities,
                    version: MessageVersion::V1,
                }
            }
        }
    */

    /*
        prop_compose! {
            fn gen_sync_message_v2()(
                heads in gen_sorted_hashes(0..10),
                need in gen_sorted_hashes(0..10),
                have in proptest::collection::vec(gen_have(), 0..10),
                raw in proptest::collection::vec(any::<u8>(), 0..100),
                supported_capabilities in prop_oneof![
                    Just(None),
                    Just(Some(vec![Capability::MessageV1])),
                    Just(Some(vec![Capability::MessageV2])),
                    Just(Some(vec![Capability::MessageV1, Capability::MessageV2])),
                ],
            ) -> Message {
                Message {
                    heads,
                    need,
                    have,
                    changes: ChunkList::from(raw),
                    supported_capabilities,
                    version: MessageVersion::V2,
                }
            }
        }
    */

    /*
        fn gen_sync_message() -> impl Strategy<Value = Message> {
            prop_oneof![gen_sync_message_v1(), gen_sync_message_v2(),].boxed()
        }
    */

    #[test]
    fn encode_decode_empty_message() {
        let msg = Message {
            heads: vec![],
            need: vec![],
            have: vec![],
            changes: ChunkList::empty(),
            flags: None,
            version: MessageVersion::V2,
        };
        let encoded = msg.encode();
        Message::parse(Input::new(&encoded)).unwrap();
    }

    /*
        proptest! {
            #[test]
            fn encode_decode_message(msg in gen_sync_message()) {
                let encoded = msg.clone().encode();
                let (i, decoded) = Message::parse(Input::new(&encoded)).unwrap();
                assert!(i.is_empty());
                assert_eq!(msg, decoded);
            }
        }
    */

    #[test]
    fn generate_sync_message_twice_does_nothing() {
        let mut doc = crate::AutoCommit::new();
        doc.put(crate::ROOT, "key", "value").unwrap();
        let mut sync_state = State::new();

        assert!(doc.sync().generate_sync_message(&mut sync_state).is_some());
        assert!(doc.sync().generate_sync_message(&mut sync_state).is_none());
    }

    #[test]
    fn first_response_is_some_even_if_no_changes() {
        // The first time we generate a sync message for a given peer we should always send a
        // response so that they know what our heads are, even if we are at the same heads as them

        let mut doc1 = crate::AutoCommit::new();
        doc1.put(crate::ROOT, "key", "value").unwrap();
        let mut doc2 = doc1.fork();

        let mut s1 = State::new();
        let mut s2 = State::new();

        let m1 = doc1
            .sync()
            .generate_sync_message(&mut s1)
            .expect("message was none");

        doc2.sync().receive_sync_message(&mut s2, m1).unwrap();

        let _m2 = doc2
            .sync()
            .generate_sync_message(&mut s2)
            .expect("response was none");
    }

    #[test]
    fn should_not_reply_if_we_have_no_data_after_first_round() {
        let mut doc1 = crate::AutoCommit::new();
        let mut doc2 = crate::AutoCommit::new();
        let mut s1 = State::new();
        let mut s2 = State::new();
        let m1 = doc1
            .sync()
            .generate_sync_message(&mut s1)
            .expect("message was none");

        doc2.sync().receive_sync_message(&mut s2, m1).unwrap();
        let _m2 = doc2
            .sync()
            .generate_sync_message(&mut s2)
            .expect("first round message was none");

        let m1 = doc1.sync().generate_sync_message(&mut s1);
        assert!(m1.is_none());

        let m2 = doc2.sync().generate_sync_message(&mut s2);
        assert!(m2.is_none());
    }

    #[test]
    fn should_allow_simultaneous_messages_during_synchronisation() {
        // create & synchronize two nodes
        let mut doc1 = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut doc2 = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());
        let mut s1 = State::new();
        let mut s2 = State::new();

        for i in 0..5 {
            doc1.put(&crate::ROOT, "x", i).unwrap();
            doc1.commit();
            doc2.put(&crate::ROOT, "y", i).unwrap();
            doc2.commit();
        }

        let head1 = doc1.get_heads()[0];
        let head2 = doc2.get_heads()[0];

        //// both sides report what they have but have no shared peer state
        let msg1to2 = doc1
            .sync()
            .generate_sync_message(&mut s1)
            .expect("initial sync from 1 to 2 was None");
        let msg2to1 = doc2
            .sync()
            .generate_sync_message(&mut s2)
            .expect("initial sync message from 2 to 1 was None");
        let Message {
            changes: changes1to2,
            ..
        } = &msg1to2;
        assert_eq!(changes1to2.len(), 0);
        assert_eq!(msg1to2.have[0].last_sync.len(), 0);
        let Message {
            changes: changes2to1,
            ..
        } = &msg2to1;
        assert_eq!(changes2to1.len(), 0);
        assert_eq!(msg2to1.have[0].last_sync.len(), 0);

        //// doc1 and doc2 receive that message and update sync state
        doc1.sync().receive_sync_message(&mut s1, msg2to1).unwrap();
        doc2.sync().receive_sync_message(&mut s2, msg1to2).unwrap();

        //// now both reply with their local changes the other lacks
        //// (standard warning that 1% of the time this will result in a "need" message)
        let msg1to2 = doc1
            .sync()
            .generate_sync_message(&mut s1)
            .expect("first reply from 1 to 2 was None");
        let Message {
            changes: changes1to2,
            ..
        } = &msg1to2;
        assert!(!changes1to2.is_empty());

        let msg2to1 = doc2
            .sync()
            .generate_sync_message(&mut s2)
            .expect("first reply from 2 to 1 was None");
        let Message {
            changes: changes2to1,
            ..
        } = &msg2to1;
        assert!(!changes2to1.is_empty());

        //// both should now apply the changes
        doc1.sync().receive_sync_message(&mut s1, msg2to1).unwrap();
        assert_eq!(doc1.get_missing_deps(&[]), Vec::new());

        doc2.sync().receive_sync_message(&mut s2, msg1to2).unwrap();
        assert_eq!(doc2.get_missing_deps(&[]), Vec::new());

        //// The response acknowledges the changes received and sends no further changes
        let msg1to2 = doc1
            .sync()
            .generate_sync_message(&mut s1)
            .expect("second reply from 1 to 2 was None");
        let Message {
            changes: changes1to2,
            ..
        } = &msg1to2;
        assert_eq!(changes1to2.len(), 0);
        let msg2to1 = doc2
            .sync()
            .generate_sync_message(&mut s2)
            .expect("second reply from 2 to 1 was None");
        let Message {
            changes: changes2to1,
            ..
        } = &msg2to1;
        assert_eq!(changes2to1.len(), 0);

        //// After receiving acknowledgements, their shared heads should be equal
        doc1.sync().receive_sync_message(&mut s1, msg2to1).unwrap();
        doc2.sync().receive_sync_message(&mut s2, msg1to2).unwrap();

        assert_eq!(s1.shared_heads, s2.shared_heads);

        //// We're in sync, no more messages required
        assert!(doc1.sync().generate_sync_message(&mut s1).is_none());
        assert!(doc2.sync().generate_sync_message(&mut s2).is_none());

        //// If we make one more change and start another sync then its lastSync should be updated
        doc1.put(crate::ROOT, "x", 5).unwrap();
        doc1.commit();
        let msg1to2 = doc1
            .sync()
            .generate_sync_message(&mut s1)
            .expect("third reply from 1 to 2 was None");
        let mut expected_heads = vec![head1, head2];
        expected_heads.sort();
        let mut actual_heads = msg1to2.have[0].last_sync.clone();
        actual_heads.sort();
        assert_eq!(actual_heads, expected_heads);
    }

    #[test]
    fn should_handle_false_positive_head() {
        // Scenario:                                                            ,-- n1
        // c0 <-- c1 <-- c2 <-- c3 <-- c4 <-- c5 <-- c6 <-- c7 <-- c8 <-- c9 <-+
        //                                                                      `-- n2
        // where n2 is a false positive in the Bloom filter containing {n1}.
        // lastSync is c9.

        let mut doc1 = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut doc2 = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());
        let mut s1 = State::new();
        let mut s2 = State::new();

        for i in 0..10 {
            doc1.put(crate::ROOT, "x", i).unwrap();
            doc1.commit();
        }

        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        // search for false positive; see comment above
        let mut i = 0;
        let (mut doc1, mut doc2) = loop {
            let mut doc1copy = doc1
                .clone()
                .with_actor(ActorId::try_from("01234567").unwrap());
            let val1 = format!("{} @ n1", i);
            doc1copy.put(crate::ROOT, "x", val1).unwrap();
            doc1copy.commit();

            let mut doc2copy = doc1
                .clone()
                .with_actor(ActorId::try_from("89abcdef").unwrap());
            let val2 = format!("{} @ n2", i);
            doc2copy.put(crate::ROOT, "x", val2).unwrap();
            doc2copy.commit();

            let n1_bloom = BloomFilter::from_hashes(doc1copy.get_heads().into_iter());
            if n1_bloom.contains_hash(&doc2copy.get_heads()[0]) {
                break (doc1copy, doc2copy);
            }
            i += 1;
        };

        let mut all_heads = doc1.get_heads();
        all_heads.extend(doc2.get_heads());
        all_heads.sort();

        // reset sync states
        let (_, mut s1) = State::parse(Input::new(s1.encode().as_slice())).unwrap();
        let (_, mut s2) = State::parse(Input::new(s2.encode().as_slice())).unwrap();
        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);
        assert_eq!(doc1.get_heads(), all_heads);
        assert_eq!(doc2.get_heads(), all_heads);
    }

    #[test]
    fn should_handle_chains_of_false_positives() {
        //// Scenario:                         ,-- c5
        //// c0 <-- c1 <-- c2 <-- c3 <-- c4 <-+
        ////                                   `-- n2c1 <-- n2c2 <-- n2c3
        //// where n2c1 and n2c2 are both false positives in the Bloom filter containing {c5}.
        //// lastSync is c4.
        let mut doc1 = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut doc2 = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());
        let mut s1 = State::new();
        let mut s2 = State::new();

        for i in 0..10 {
            doc1.put(crate::ROOT, "x", i).unwrap();
            doc1.commit();
        }

        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        doc1.put(crate::ROOT, "x", 5).unwrap();
        doc1.commit();
        let bloom = BloomFilter::from_hashes(doc1.get_heads().into_iter());

        // search for false positive; see comment above
        let mut i = 0;
        let mut doc2 = loop {
            let mut doc = doc2
                .fork()
                .with_actor(ActorId::try_from("89abcdef").unwrap());
            doc.put(crate::ROOT, "x", format!("{} at 89abdef", i))
                .unwrap();
            doc.commit();
            if bloom.contains_hash(&doc.get_heads()[0]) {
                break doc;
            }
            i += 1;
        };

        // find another false positive building on the first
        i = 0;
        let mut doc2 = loop {
            let mut doc = doc2
                .fork()
                .with_actor(ActorId::try_from("89abcdef").unwrap());
            doc.put(crate::ROOT, "x", format!("{} again", i)).unwrap();
            doc.commit();
            if bloom.contains_hash(&doc.get_heads()[0]) {
                break doc;
            }
            i += 1;
        };

        doc2.put(crate::ROOT, "x", "final @ 89abcdef").unwrap();

        let mut all_heads = doc1.get_heads();
        all_heads.extend(doc2.get_heads());
        all_heads.sort();

        let (_, mut s1) = State::parse(Input::new(s1.encode().as_slice())).unwrap();
        let (_, mut s2) = State::parse(Input::new(s2.encode().as_slice())).unwrap();
        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);
        assert_eq!(doc1.get_heads(), all_heads);
        assert_eq!(doc2.get_heads(), all_heads);
    }

    #[test]
    fn should_handle_lots_of_branching_and_merging() {
        let mut doc1 = crate::AutoCommit::new().with_actor(ActorId::try_from("01234567").unwrap());
        let mut doc2 = crate::AutoCommit::new().with_actor(ActorId::try_from("89abcdef").unwrap());
        let mut doc3 = crate::AutoCommit::new().with_actor(ActorId::try_from("fedcba98").unwrap());
        let mut s1 = State::new();
        let mut s2 = State::new();

        doc1.put(crate::ROOT, "x", 0).unwrap();
        let change1 = doc1.get_last_local_change().unwrap().clone();

        doc2.apply_changes([change1.clone()]).unwrap();
        doc3.apply_changes([change1]).unwrap();

        doc3.put(crate::ROOT, "x", 1).unwrap();

        ////        - n1c1 <------ n1c2 <------ n1c3 <-- etc. <-- n1c20 <------ n1c21
        ////       /          \/           \/                              \/
        ////      /           /\           /\                              /\
        //// c0 <---- n2c1 <------ n2c2 <------ n2c3 <-- etc. <-- n2c20 <------ n2c21
        ////      \                                                          /
        ////       ---------------------------------------------- n3c1 <-----
        for i in 1..20 {
            doc1.put(crate::ROOT, "n1", i).unwrap();
            doc2.put(crate::ROOT, "n2", i).unwrap();
            let change1 = doc1.get_last_local_change().unwrap().clone();
            let change2 = doc2.get_last_local_change().unwrap().clone();
            doc1.apply_changes([change2.clone()]).unwrap();
            doc2.apply_changes([change1]).unwrap();
        }

        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        //// Having n3's last change concurrent to the last sync heads forces us into the slower code path
        let change3 = doc3.get_last_local_change().unwrap().clone();
        doc2.apply_changes([change3]).unwrap();

        doc1.put(crate::ROOT, "n1", "final").unwrap();
        doc2.put(crate::ROOT, "n1", "final").unwrap();

        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        assert_eq!(doc1.get_heads(), doc2.get_heads());
    }

    #[test]
    fn in_flight_logic_should_not_sabotage_concurrent_changes() {
        // This reproduces issue https://github.com/automerge/automerge/issues/702
        //
        // This problem manifested as a situation where the sync states of two
        // ends of a connection both return None from `generate_sync_message` -
        // indicating that there is nothing to send - yet the documents were
        // different at either end.

        // Because this logic depends on bloom filter false positives we have to
        // run the test many times, hence this loop
        for _ in 0..300 {
            // create two documents
            let mut doc1 = crate::AutoCommit::new();
            let mut doc2 = crate::AutoCommit::new();
            let mut s1 = State::new();
            let mut s2 = State::new();

            // get them in sync
            sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

            // make a change on doc2
            doc2.put(crate::ROOT, "x", 0).unwrap();

            // generate a sync message containing the change (this should
            // alwasy be Some because we have generated new local changes)
            let msg = doc2.sync().generate_sync_message(&mut s2).unwrap();
            // Receive that sync message on doc1
            doc1.sync().receive_sync_message(&mut s1, msg).unwrap();

            // now before sending any messages back to doc2, make a change on
            // doc1
            doc1.put(crate::ROOT, "x", 1).unwrap();

            // now synchronize
            sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

            // At this point both documents should be equal
            assert_eq!(doc1.get_heads(), doc2.get_heads());
        }
    }

    fn sync(
        a: &mut crate::AutoCommit,
        b: &mut crate::AutoCommit,
        a_sync_state: &mut State,
        b_sync_state: &mut State,
    ) {
        //function sync(a: Automerge, b: Automerge, aSyncState = initSyncState(), bSyncState = initSyncState()) {
        const MAX_ITER: usize = 10;
        let mut iterations = 0;

        loop {
            let a_to_b = a.sync().generate_sync_message(a_sync_state);
            let b_to_a = b.sync().generate_sync_message(b_sync_state);
            if a_to_b.is_none() && b_to_a.is_none() {
                break;
            }
            if iterations > MAX_ITER {
                panic!("failed to sync in {} iterations", MAX_ITER);
            }
            if let Some(msg) = a_to_b {
                b.sync().receive_sync_message(b_sync_state, msg).unwrap()
            }
            if let Some(msg) = b_to_a {
                a.sync().receive_sync_message(a_sync_state, msg).unwrap()
            }
            iterations += 1;
        }
    }

    #[test]
    fn if_first_message_has_no_heads_and_supports_v2_message_send_whole_doc() {
        let mut doc1 = crate::AutoCommit::new();
        let mut doc2 = crate::AutoCommit::new();
        doc2.put(crate::ROOT, "foo", "bar").unwrap();

        let mut s1 = State::new();
        let mut s2 = State::new();

        let outgoing = doc1
            .sync()
            .generate_sync_message(&mut s1)
            .expect("message was none");

        doc2.sync().receive_sync_message(&mut s2, outgoing).unwrap();

        let response = doc2
            .sync()
            .generate_sync_message(&mut s2)
            .expect("response was none");

        let Message { changes, .. } = response;

        let (_, chunk) = Chunk::parse(Input::new(&changes.0[0])).unwrap();
        assert!(matches!(chunk, Chunk::Document(_)));
    }

    #[test]
    fn read_only_sync_does_not_apply_incoming_changes() {
        let mut doc1 = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut doc2 = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        doc1.put(crate::ROOT, "from_doc1", "hello").unwrap();
        doc1.commit();
        doc2.put(crate::ROOT, "from_doc2", "world").unwrap();
        doc2.commit();

        let doc1_heads_before = doc1.get_heads();
        let doc2_heads_before = doc2.get_heads();

        // doc1 is read-only: it should send its changes but not accept doc2's
        let mut s1 = State::new_read_only();
        let mut s2 = State::new();

        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        // doc2 should have received doc1's changes
        assert!(doc2.get(crate::ROOT, "from_doc1").unwrap().is_some());
        assert!(doc2.get(crate::ROOT, "from_doc2").unwrap().is_some());

        // doc1 should NOT have received doc2's changes
        assert!(doc1.get(crate::ROOT, "from_doc1").unwrap().is_some());
        assert!(doc1.get(crate::ROOT, "from_doc2").unwrap().is_none());

        // doc1's heads should be unchanged
        assert_eq!(doc1.get_heads(), doc1_heads_before);

        // doc2's heads should have advanced
        assert_ne!(doc2.get_heads(), doc2_heads_before);
    }

    #[test]
    fn read_only_empty_peer_syncs_with_data_peer() {
        // An empty read-only peer syncs with a peer that has data.
        // This exercises the V2 "send full document" path on the non-read-only side.
        // The read-only peer should ignore the document and the protocol should converge.
        let mut doc1 = crate::AutoCommit::new();
        let mut doc2 = crate::AutoCommit::new();
        doc2.put(crate::ROOT, "key", "value").unwrap();
        doc2.commit();

        let mut s1 = State::new_read_only();
        let mut s2 = State::new();

        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        // doc1 should still be empty
        assert!(doc1.get(crate::ROOT, "key").unwrap().is_none());
        assert!(doc1.get_heads().is_empty());

        // doc2 should be unchanged
        assert!(doc2.get(crate::ROOT, "key").unwrap().is_some());
    }

    #[test]
    fn both_peers_read_only() {
        // When both peers are read-only, neither applies the other's changes.
        // The protocol should still converge.
        let mut doc1 = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut doc2 = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        doc1.put(crate::ROOT, "from_doc1", "hello").unwrap();
        doc1.commit();
        doc2.put(crate::ROOT, "from_doc2", "world").unwrap();
        doc2.commit();

        let doc1_heads = doc1.get_heads();
        let doc2_heads = doc2.get_heads();

        let mut s1 = State::new_read_only();
        let mut s2 = State::new_read_only();

        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        // Neither peer should have the other's changes
        assert!(doc1.get(crate::ROOT, "from_doc2").unwrap().is_none());
        assert!(doc2.get(crate::ROOT, "from_doc1").unwrap().is_none());

        // Both heads unchanged
        assert_eq!(doc1.get_heads(), doc1_heads);
        assert_eq!(doc2.get_heads(), doc2_heads);
    }

    #[test]
    fn both_peers_read_only_converges_to_none() {
        // Explicitly verify that generate_sync_message returns None on both
        // sides after the initial exchange between two read-only peers.
        let mut doc1 = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut doc2 = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        doc1.put(crate::ROOT, "from_doc1", "hello").unwrap();
        doc1.commit();
        doc2.put(crate::ROOT, "from_doc2", "world").unwrap();
        doc2.commit();

        let mut s1 = State::new_read_only();
        let mut s2 = State::new_read_only();

        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        // After sync, both must return None — no infinite loop
        assert!(doc1.sync().generate_sync_message(&mut s1).is_none());
        assert!(doc2.sync().generate_sync_message(&mut s2).is_none());

        // Both discover the other is read-only
        assert!(s1.is_peer_read_only());
        assert!(s2.is_peer_read_only());
    }

    #[test]
    fn both_read_only_one_makes_local_changes() {
        // Both peers are read-only. One makes local changes between sync
        // rounds. The updated heads should be communicated (no changes sent),
        // and the protocol should converge.
        let mut doc1 = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut doc2 = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        let mut s1 = State::new_read_only();
        let mut s2 = State::new_read_only();

        // Initial sync with no data — just exchange hellos
        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        // doc1 makes local changes
        doc1.put(crate::ROOT, "key", "value1").unwrap();
        doc1.commit();
        let doc1_heads_after = doc1.get_heads();

        // Sync again — should converge, doc1's new heads communicated
        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        // doc2 should know about doc1's heads but not have the changes
        assert_eq!(s2.their_heads.as_ref().unwrap(), &doc1_heads_after);
        assert!(doc2.get(crate::ROOT, "key").unwrap().is_none());

        // doc1 makes more changes
        doc1.put(crate::ROOT, "key", "value2").unwrap();
        doc1.commit();
        let doc1_heads_after2 = doc1.get_heads();

        // Sync again
        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        // doc2 should see the updated heads
        assert_eq!(s2.their_heads.as_ref().unwrap(), &doc1_heads_after2);
        assert!(doc2.get(crate::ROOT, "key").unwrap().is_none());

        // Must converge
        assert!(doc1.sync().generate_sync_message(&mut s1).is_none());
        assert!(doc2.sync().generate_sync_message(&mut s2).is_none());
    }

    #[test]
    fn both_read_only_both_make_local_changes() {
        // Both peers are read-only and both make local changes between sync
        // rounds. Both should learn the other's updated heads, neither should
        // receive actual changes, and the protocol should converge each round.
        let mut doc1 = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut doc2 = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        let mut s1 = State::new_read_only();
        let mut s2 = State::new_read_only();

        for round in 0..5 {
            doc1.put(crate::ROOT, "doc1_counter", round as i64).unwrap();
            doc1.commit();
            doc2.put(crate::ROOT, "doc2_counter", round as i64).unwrap();
            doc2.commit();

            let doc1_heads = doc1.get_heads();
            let doc2_heads = doc2.get_heads();

            // Must converge each round (sync helper panics after 10 iterations)
            sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

            // Each peer should know the other's heads
            assert_eq!(
                s1.their_heads.as_ref().unwrap(),
                &doc2_heads,
                "round {round}: doc1 should know doc2's heads"
            );
            assert_eq!(
                s2.their_heads.as_ref().unwrap(),
                &doc1_heads,
                "round {round}: doc2 should know doc1's heads"
            );

            // Neither should have the other's data
            assert!(
                doc1.get(crate::ROOT, "doc2_counter").unwrap().is_none(),
                "round {round}: doc1 should not have doc2's changes"
            );
            assert!(
                doc2.get(crate::ROOT, "doc1_counter").unwrap().is_none(),
                "round {round}: doc2 should not have doc1's changes"
            );

            // Must be fully converged
            assert!(
                doc1.sync().generate_sync_message(&mut s1).is_none(),
                "round {round}: doc1 should have nothing more to send"
            );
            assert!(
                doc2.sync().generate_sync_message(&mut s2).is_none(),
                "round {round}: doc2 should have nothing more to send"
            );
        }
    }

    #[test]
    fn both_read_only_simultaneous_changes_during_sync() {
        // Both peers are read-only. Both make changes, then exchange messages
        // simultaneously (like the simultaneous sync test). Must converge.
        for _ in 0..100 {
            let mut doc1 =
                crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
            let mut doc2 =
                crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

            let mut s1 = State::new_read_only();
            let mut s2 = State::new_read_only();

            // Both make changes
            doc1.put(crate::ROOT, "x", 1).unwrap();
            doc1.commit();
            doc2.put(crate::ROOT, "y", 2).unwrap();
            doc2.commit();

            // Sync (uses the helper which panics on non-convergence)
            sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

            // Now both make MORE changes
            doc1.put(crate::ROOT, "x", 3).unwrap();
            doc1.commit();
            doc2.put(crate::ROOT, "y", 4).unwrap();
            doc2.commit();

            // Sync again
            sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

            // Must be fully converged
            assert!(doc1.sync().generate_sync_message(&mut s1).is_none());
            assert!(doc2.sync().generate_sync_message(&mut s2).is_none());

            // Neither has the other's data
            assert!(doc1.get(crate::ROOT, "y").unwrap().is_none());
            assert!(doc2.get(crate::ROOT, "x").unwrap().is_none());
        }
    }

    #[test]
    fn read_only_peer_new_changes_between_sync_rounds() {
        // After an initial sync converges, the read-only peer makes new local
        // changes and syncs again. The new changes should flow to the other peer.
        let mut doc1 = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut doc2 = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        doc1.put(crate::ROOT, "round1", "from_doc1").unwrap();
        doc1.commit();
        doc2.put(crate::ROOT, "round1", "from_doc2").unwrap();
        doc2.commit();

        let mut s1 = State::new_read_only();
        let mut s2 = State::new();

        // First sync round
        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);
        assert!(doc2.get(crate::ROOT, "round1").unwrap().is_some());

        // doc1 makes new changes
        doc1.put(crate::ROOT, "round2", "new_from_doc1").unwrap();
        doc1.commit();

        // doc2 also makes new changes
        doc2.put(crate::ROOT, "round2", "new_from_doc2").unwrap();
        doc2.commit();

        // Second sync round
        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        // doc2 should have doc1's new changes
        assert_eq!(
            doc2.get(crate::ROOT, "round2").unwrap().unwrap().0.to_str(),
            // doc2 has both values as a conflict, but the winning value depends on actor ordering;
            // just check the key exists
            doc2.get(crate::ROOT, "round2").unwrap().unwrap().0.to_str(),
        );
        // Verify doc2 has received "new_from_doc1" somewhere in the conflicts
        let all_values: Vec<_> = doc2
            .get_all(crate::ROOT, "round2")
            .unwrap()
            .into_iter()
            .map(|(v, _)| v.into_string().unwrap())
            .collect();
        assert!(all_values.contains(&"new_from_doc1".to_string()));
        assert!(all_values.contains(&"new_from_doc2".to_string()));

        // doc1 should still NOT have doc2's changes
        assert!(
            doc1.get(crate::ROOT, "from_doc2").is_err()
                || doc1.get(crate::ROOT, "from_doc2").unwrap().is_none()
        );
        let doc1_round2_values: Vec<_> = doc1
            .get_all(crate::ROOT, "round2")
            .unwrap()
            .into_iter()
            .map(|(v, _)| v.into_string().unwrap())
            .collect();
        assert_eq!(doc1_round2_values, vec!["new_from_doc1".to_string()]);
    }

    #[test]
    fn read_only_peer_concurrent_changes_during_sync() {
        // Like in_flight_logic_should_not_sabotage_concurrent_changes but with
        // a read-only peer. The read-only peer makes a change after receiving a
        // message but before the sync loop completes.
        for _ in 0..300 {
            let mut doc1 =
                crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
            let mut doc2 =
                crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());
            let mut s1 = State::new_read_only();
            let mut s2 = State::new();

            // get them in sync (doc1 read-only, so only doc2 gets doc1's changes)
            sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

            // doc2 makes a change
            doc2.put(crate::ROOT, "x", 0).unwrap();

            // generate + receive one message from doc2 to doc1
            let msg = doc2.sync().generate_sync_message(&mut s2).unwrap();
            doc1.sync().receive_sync_message(&mut s1, msg).unwrap();

            // before sending anything back, doc1 (read-only) makes a local change
            doc1.put(crate::ROOT, "y", 1).unwrap();

            // now synchronize the rest
            sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

            // doc2 should have doc1's change
            assert!(doc2.get(crate::ROOT, "y").unwrap().is_some());
            // doc1 should NOT have doc2's change
            assert!(doc1.get(crate::ROOT, "x").unwrap().is_none());
        }
    }

    #[test]
    fn read_only_publisher_to_multiple_consumers() {
        // R (read-only) publishes to A and B separately.
        // A makes its own changes. Then B syncs with R again.
        // B should NOT get A's changes through R (R never accepted them).
        let mut r = crate::AutoCommit::new().with_actor(ActorId::try_from("aaaaaa").unwrap());
        let mut a = crate::AutoCommit::new().with_actor(ActorId::try_from("bbbbbb").unwrap());
        let mut b = crate::AutoCommit::new().with_actor(ActorId::try_from("cccccc").unwrap());

        r.put(crate::ROOT, "from_r", "hello").unwrap();
        r.commit();

        // R syncs with A (read-only: A gets R's changes, R ignores A's)
        let mut sr_a = State::new_read_only();
        let mut sa_r = State::new();
        sync(&mut r, &mut a, &mut sr_a, &mut sa_r);
        assert!(a.get(crate::ROOT, "from_r").unwrap().is_some());

        // A makes its own changes
        a.put(crate::ROOT, "from_a", "world").unwrap();
        a.commit();

        // A syncs with R — R should ignore A's changes
        sync(&mut r, &mut a, &mut sr_a, &mut sa_r);
        assert!(r.get(crate::ROOT, "from_a").unwrap().is_none());

        // R syncs with B (read-only: B gets R's changes, R ignores B's)
        let mut sr_b = State::new_read_only();
        let mut sb_r = State::new();
        sync(&mut r, &mut b, &mut sr_b, &mut sb_r);

        // B should have R's changes but NOT A's
        assert!(b.get(crate::ROOT, "from_r").unwrap().is_some());
        assert!(b.get(crate::ROOT, "from_a").unwrap().is_none());
    }

    #[test]
    fn triangle_changes_arrive_via_two_paths() {
        // R (read-only) makes changes, syncs with A. A gets R's changes.
        // A syncs with B (read-write). B gets R's changes via A.
        // Now B syncs with R directly. R should recognize B already has its
        // changes (via bloom filter) and not redundantly send them.
        // The protocol should converge.
        let mut r = crate::AutoCommit::new().with_actor(ActorId::try_from("aaaaaa").unwrap());
        let mut a = crate::AutoCommit::new().with_actor(ActorId::try_from("bbbbbb").unwrap());
        let mut b = crate::AutoCommit::new().with_actor(ActorId::try_from("cccccc").unwrap());

        r.put(crate::ROOT, "from_r", "hello").unwrap();
        r.commit();
        a.put(crate::ROOT, "from_a", "world").unwrap();
        a.commit();

        // R syncs with A (read-only)
        let mut sr_a = State::new_read_only();
        let mut sa_r = State::new();
        sync(&mut r, &mut a, &mut sr_a, &mut sa_r);

        // A now has R's changes
        assert!(a.get(crate::ROOT, "from_r").unwrap().is_some());

        // A syncs with B (both read-write)
        let mut sa_b = State::new();
        let mut sb_a = State::new();
        sync(&mut a, &mut b, &mut sa_b, &mut sb_a);

        // B now has R's changes (received via A)
        assert!(b.get(crate::ROOT, "from_r").unwrap().is_some());
        assert!(b.get(crate::ROOT, "from_a").unwrap().is_some());

        // Now B syncs with R directly
        let mut sr_b = State::new_read_only();
        let mut sb_r = State::new();
        sync(&mut r, &mut b, &mut sr_b, &mut sb_r);

        // R should still only have its own changes
        assert!(r.get(crate::ROOT, "from_a").unwrap().is_none());

        // B should still have everything (R's changes aren't duplicated, just confirmed)
        assert!(b.get(crate::ROOT, "from_r").unwrap().is_some());
        assert!(b.get(crate::ROOT, "from_a").unwrap().is_some());
    }

    #[test]
    fn read_only_fully_connected_triangle() {
        // Three peers: A, B, R. R is read-only toward both A and B.
        // All three have their own changes. Each pair syncs.
        // A and B should converge to having A's + B's + R's changes.
        // R should only have its own changes.
        let mut r = crate::AutoCommit::new().with_actor(ActorId::try_from("aaaaaa").unwrap());
        let mut a = crate::AutoCommit::new().with_actor(ActorId::try_from("bbbbbb").unwrap());
        let mut b = crate::AutoCommit::new().with_actor(ActorId::try_from("cccccc").unwrap());

        r.put(crate::ROOT, "from_r", "r_val").unwrap();
        r.commit();
        a.put(crate::ROOT, "from_a", "a_val").unwrap();
        a.commit();
        b.put(crate::ROOT, "from_b", "b_val").unwrap();
        b.commit();

        let r_heads = r.get_heads();

        // R syncs with A (read-only)
        let mut sr_a = State::new_read_only();
        let mut sa_r = State::new();
        sync(&mut r, &mut a, &mut sr_a, &mut sa_r);

        // R syncs with B (read-only)
        let mut sr_b = State::new_read_only();
        let mut sb_r = State::new();
        sync(&mut r, &mut b, &mut sr_b, &mut sb_r);

        // A and B each have R's changes via separate sync sessions
        assert!(a.get(crate::ROOT, "from_r").unwrap().is_some());
        assert!(b.get(crate::ROOT, "from_r").unwrap().is_some());

        // A syncs with B (read-write) — both already have R's changes
        let mut sa_b = State::new();
        let mut sb_a = State::new();
        sync(&mut a, &mut b, &mut sa_b, &mut sb_a);

        // A and B should have all three sets of changes
        assert!(a.get(crate::ROOT, "from_a").unwrap().is_some());
        assert!(a.get(crate::ROOT, "from_b").unwrap().is_some());
        assert!(a.get(crate::ROOT, "from_r").unwrap().is_some());
        assert!(b.get(crate::ROOT, "from_a").unwrap().is_some());
        assert!(b.get(crate::ROOT, "from_b").unwrap().is_some());
        assert!(b.get(crate::ROOT, "from_r").unwrap().is_some());

        // A and B should have the same heads
        assert_eq!(a.get_heads(), b.get_heads());

        // R should only have its own changes
        assert_eq!(r.get_heads(), r_heads);
        assert!(r.get(crate::ROOT, "from_a").unwrap().is_none());
        assert!(r.get(crate::ROOT, "from_b").unwrap().is_none());
    }

    #[test]
    fn stale_shared_heads_after_read_only_sync() {
        // R (read-only) syncs with A. Then A syncs with B (B gets everything).
        // Now B syncs with R using a fresh State. R tries to send its changes
        // to B, but B already has them (received via A). The bloom filter
        // should prevent redundant sending and the protocol should converge.
        let mut r = crate::AutoCommit::new().with_actor(ActorId::try_from("aaaaaa").unwrap());
        let mut a = crate::AutoCommit::new().with_actor(ActorId::try_from("bbbbbb").unwrap());
        let mut b = crate::AutoCommit::new().with_actor(ActorId::try_from("cccccc").unwrap());

        // R has several changes to make bloom filter interaction interesting
        for i in 0..10 {
            r.put(crate::ROOT, "counter", i as i64).unwrap();
            r.commit();
        }
        a.put(crate::ROOT, "from_a", "a_val").unwrap();
        a.commit();

        // R syncs with A (read-only)
        let mut sr_a = State::new_read_only();
        let mut sa_r = State::new();
        sync(&mut r, &mut a, &mut sr_a, &mut sa_r);

        // A now has R's 10 changes
        assert!(a.get(crate::ROOT, "counter").unwrap().is_some());

        // A syncs with B — B gets everything (R's changes + A's changes)
        let mut sa_b = State::new();
        let mut sb_a = State::new();
        sync(&mut a, &mut b, &mut sa_b, &mut sb_a);

        assert!(b.get(crate::ROOT, "counter").unwrap().is_some());
        assert!(b.get(crate::ROOT, "from_a").unwrap().is_some());

        // Now B syncs directly with R (fresh sync state)
        let mut sr_b = State::new_read_only();
        let mut sb_r = State::new();
        sync(&mut r, &mut b, &mut sr_b, &mut sb_r);

        // R should not have B's or A's changes
        assert!(r.get(crate::ROOT, "from_a").unwrap().is_none());

        // B should still have everything
        assert!(b.get(crate::ROOT, "counter").unwrap().is_some());
        assert!(b.get(crate::ROOT, "from_a").unwrap().is_some());
    }

    #[test]
    fn read_only_peer_receives_same_changes_from_two_peers() {
        // A and B sync with each other first (both have the same data).
        // Then A syncs with R (read-only), then B syncs with R (read-only).
        // R ignores changes both times. R's sync state must handle receiving
        // announcements about the same changes from two different peers.
        let mut r = crate::AutoCommit::new().with_actor(ActorId::try_from("aaaaaa").unwrap());
        let mut a = crate::AutoCommit::new().with_actor(ActorId::try_from("bbbbbb").unwrap());
        let mut b = crate::AutoCommit::new().with_actor(ActorId::try_from("cccccc").unwrap());

        r.put(crate::ROOT, "from_r", "r_val").unwrap();
        r.commit();
        a.put(crate::ROOT, "from_a", "a_val").unwrap();
        a.commit();
        b.put(crate::ROOT, "from_b", "b_val").unwrap();
        b.commit();

        // A and B sync — they now share A's + B's changes
        let mut sa_b = State::new();
        let mut sb_a = State::new();
        sync(&mut a, &mut b, &mut sa_b, &mut sb_a);
        assert_eq!(a.get_heads(), b.get_heads());

        let r_heads = r.get_heads();

        // A syncs with R (read-only) — R ignores A's+B's changes, A gets R's
        let mut sr_a = State::new_read_only();
        let mut sa_r = State::new();
        sync(&mut r, &mut a, &mut sr_a, &mut sa_r);
        assert!(a.get(crate::ROOT, "from_r").unwrap().is_some());
        assert_eq!(r.get_heads(), r_heads);

        // B syncs with R (read-only) — R ignores the same changes again, B gets R's
        let mut sr_b = State::new_read_only();
        let mut sb_r = State::new();
        sync(&mut r, &mut b, &mut sr_b, &mut sb_r);
        assert!(b.get(crate::ROOT, "from_r").unwrap().is_some());

        // R should still only have its own changes
        assert_eq!(r.get_heads(), r_heads);
        assert!(r.get(crate::ROOT, "from_a").unwrap().is_none());
        assert!(r.get(crate::ROOT, "from_b").unwrap().is_none());

        // R makes a new change
        r.put(crate::ROOT, "from_r_2", "new").unwrap();
        r.commit();

        // Sync R with A again — A should get the new change
        sync(&mut r, &mut a, &mut sr_a, &mut sa_r);
        assert!(a.get(crate::ROOT, "from_r_2").unwrap().is_some());

        // Sync R with B again — B should also get the new change
        sync(&mut r, &mut b, &mut sr_b, &mut sb_r);
        assert!(b.get(crate::ROOT, "from_r_2").unwrap().is_some());
    }

    #[test]
    fn switch_read_only_to_read_write_mid_session() {
        // A starts read-only, syncs with B (A sends changes, ignores B's).
        // Then A switches to read-write and syncs again.
        // A should now receive B's changes despite B having previously sent
        // them (the sent_hashes problem).
        let mut a = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut b = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        a.put(crate::ROOT, "from_a", "hello").unwrap();
        a.commit();
        b.put(crate::ROOT, "from_b", "world").unwrap();
        b.commit();

        let mut sa = State::new_read_only();
        let mut sb = State::new();

        // First sync: A is read-only
        sync(&mut a, &mut b, &mut sa, &mut sb);
        assert!(b.get(crate::ROOT, "from_a").unwrap().is_some());
        assert!(a.get(crate::ROOT, "from_b").unwrap().is_none());

        // Switch A to read-write
        sa.set_read_only(false);

        // Second sync: A should now receive B's changes
        sync(&mut a, &mut b, &mut sa, &mut sb);
        assert!(a.get(crate::ROOT, "from_b").unwrap().is_some());

        // Both should have the same heads now
        assert_eq!(a.get_heads(), b.get_heads());
    }

    #[test]
    fn switch_read_write_to_read_only_mid_session() {
        // A and B sync normally (both read-write). Then A switches to
        // read-only. B makes new changes. A should not receive them.
        let mut a = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut b = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        a.put(crate::ROOT, "from_a", "hello").unwrap();
        a.commit();
        b.put(crate::ROOT, "from_b", "world").unwrap();
        b.commit();

        let mut sa = State::new();
        let mut sb = State::new();

        // First sync: both read-write
        sync(&mut a, &mut b, &mut sa, &mut sb);
        assert_eq!(a.get_heads(), b.get_heads());

        // Switch A to read-only
        sa.set_read_only(true);

        // B makes new changes
        b.put(crate::ROOT, "new_from_b", "secret").unwrap();
        b.commit();

        // A also makes a new change
        a.put(crate::ROOT, "new_from_a", "published").unwrap();
        a.commit();

        // Second sync: A is now read-only
        sync(&mut a, &mut b, &mut sa, &mut sb);

        // B should have A's new change
        assert!(b.get(crate::ROOT, "new_from_a").unwrap().is_some());
        // A should NOT have B's new change
        assert!(a.get(crate::ROOT, "new_from_b").unwrap().is_none());
    }

    #[test]
    fn switch_read_only_to_read_write_with_multiple_rounds() {
        // A starts read-only, syncs multiple rounds with B (B makes changes
        // each round). A switches to read-write. A should receive ALL of B's
        // accumulated changes.
        let mut a = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut b = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        a.put(crate::ROOT, "from_a", "initial").unwrap();
        a.commit();

        let mut sa = State::new_read_only();
        let mut sb = State::new();

        // Round 1: B has changes, A is read-only
        b.put(crate::ROOT, "round1", "from_b").unwrap();
        b.commit();
        sync(&mut a, &mut b, &mut sa, &mut sb);
        assert!(a.get(crate::ROOT, "round1").unwrap().is_none());

        // Round 2: B has more changes, A still read-only
        b.put(crate::ROOT, "round2", "from_b").unwrap();
        b.commit();
        sync(&mut a, &mut b, &mut sa, &mut sb);
        assert!(a.get(crate::ROOT, "round2").unwrap().is_none());

        // Round 3: B has even more changes, A still read-only
        b.put(crate::ROOT, "round3", "from_b").unwrap();
        b.commit();
        sync(&mut a, &mut b, &mut sa, &mut sb);
        assert!(a.get(crate::ROOT, "round3").unwrap().is_none());

        // Switch A to read-write
        sa.set_read_only(false);

        // Sync again — A should receive ALL of B's changes from all rounds
        sync(&mut a, &mut b, &mut sa, &mut sb);
        assert!(a.get(crate::ROOT, "round1").unwrap().is_some());
        assert!(a.get(crate::ROOT, "round2").unwrap().is_some());
        assert!(a.get(crate::ROOT, "round3").unwrap().is_some());

        assert_eq!(a.get_heads(), b.get_heads());
    }

    #[test]
    fn toggle_read_only_multiple_times() {
        // Rapidly toggle read-only on and off, making changes between each toggle.
        let mut a = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut b = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        let mut sa = State::new_read_only();
        let mut sb = State::new();

        // Round 1: A read-only, B makes changes
        b.put(crate::ROOT, "b1", "val").unwrap();
        b.commit();
        a.put(crate::ROOT, "a1", "val").unwrap();
        a.commit();
        sync(&mut a, &mut b, &mut sa, &mut sb);
        assert!(b.get(crate::ROOT, "a1").unwrap().is_some());
        assert!(a.get(crate::ROOT, "b1").unwrap().is_none());

        // Round 2: switch to read-write
        sa.set_read_only(false);
        b.put(crate::ROOT, "b2", "val").unwrap();
        b.commit();
        a.put(crate::ROOT, "a2", "val").unwrap();
        a.commit();
        sync(&mut a, &mut b, &mut sa, &mut sb);
        assert!(a.get(crate::ROOT, "b1").unwrap().is_some());
        assert!(a.get(crate::ROOT, "b2").unwrap().is_some());
        assert!(b.get(crate::ROOT, "a2").unwrap().is_some());

        // Round 3: switch back to read-only
        sa.set_read_only(true);
        b.put(crate::ROOT, "b3", "val").unwrap();
        b.commit();
        a.put(crate::ROOT, "a3", "val").unwrap();
        a.commit();
        sync(&mut a, &mut b, &mut sa, &mut sb);
        assert!(b.get(crate::ROOT, "a3").unwrap().is_some());
        assert!(a.get(crate::ROOT, "b3").unwrap().is_none());

        // Round 4: switch to read-write again
        sa.set_read_only(false);
        sync(&mut a, &mut b, &mut sa, &mut sb);
        assert!(a.get(crate::ROOT, "b3").unwrap().is_some());
        assert_eq!(a.get_heads(), b.get_heads());
    }

    #[test]
    fn peer_discovers_remote_read_only_status() {
        // After exchanging messages, B should discover that A is read-only.
        let mut a = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut b = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        a.put(crate::ROOT, "from_a", "hello").unwrap();
        a.commit();
        b.put(crate::ROOT, "from_b", "world").unwrap();
        b.commit();

        let mut sa = State::new_read_only();
        let mut sb = State::new();

        // Before any exchange, B doesn't know A's status
        assert!(!sb.is_peer_read_only());

        // A generates a message (includes ReadOnly capability)
        let msg = a.sync().generate_sync_message(&mut sa).unwrap();

        // B receives it and discovers A is read-only
        b.sync().receive_sync_message(&mut sb, msg).unwrap();
        assert!(sb.is_peer_read_only());

        // Complete the sync
        sync(&mut a, &mut b, &mut sa, &mut sb);
        assert!(sb.is_peer_read_only());

        // A switches to read-write
        sa.set_read_only(false);

        // After exchanging messages, B discovers A is no longer read-only
        let msg = a.sync().generate_sync_message(&mut sa).unwrap();
        b.sync().receive_sync_message(&mut sb, msg).unwrap();
        assert!(!sb.is_peer_read_only());
    }

    #[test]
    fn changes_not_sent_to_read_only_peer() {
        // B should not send changes to A when B knows A is read-only.
        // This saves bandwidth.
        let mut a = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut b = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        b.put(crate::ROOT, "from_b", "world").unwrap();
        b.commit();

        let mut sa = State::new_read_only();
        let mut sb = State::new();

        // Exchange initial messages so B discovers A is read-only
        let msg_a = a.sync().generate_sync_message(&mut sa).unwrap();
        b.sync().receive_sync_message(&mut sb, msg_a).unwrap();
        assert!(sb.is_peer_read_only());

        // B generates a response — should have no changes since A is read-only
        let msg_b = b.sync().generate_sync_message(&mut sb).unwrap();
        assert!(msg_b.changes.is_empty());

        // Complete the sync
        a.sync().receive_sync_message(&mut sa, msg_b).unwrap();
        sync(&mut a, &mut b, &mut sa, &mut sb);

        // A still doesn't have B's changes (read-only + B didn't even send them)
        assert!(a.get(crate::ROOT, "from_b").unwrap().is_none());

        // Now A switches to read-write
        sa.set_read_only(false);
        sync(&mut a, &mut b, &mut sa, &mut sb);

        // A should now have B's changes
        assert!(a.get(crate::ROOT, "from_b").unwrap().is_some());
        assert_eq!(a.get_heads(), b.get_heads());
    }

    #[test]
    fn generate_message_after_set_read_only_even_with_in_flight() {
        // After calling generate_sync_message (setting in_flight=true),
        // then set_read_only, the next generate_sync_message must still
        // produce a message so the peer learns about the mode change.
        let mut a = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut b = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        a.put(crate::ROOT, "from_a", "hello").unwrap();
        a.commit();
        b.put(crate::ROOT, "from_b", "world").unwrap();
        b.commit();

        let mut sa = State::new();
        let mut sb = State::new();

        // Normal sync to get in sync
        sync(&mut a, &mut b, &mut sa, &mut sb);

        // B makes a new change and sends it to A
        b.put(crate::ROOT, "new_from_b", "secret").unwrap();
        b.commit();
        let msg_b = b.sync().generate_sync_message(&mut sb).unwrap();
        a.sync().receive_sync_message(&mut sa, msg_b).unwrap();

        // A generates a response (sets in_flight=true)
        let msg = a.sync().generate_sync_message(&mut sa);
        assert!(msg.is_some());
        assert!(sa.in_flight);

        // A switches to read-only BEFORE receiving B's next message
        sa.set_read_only(true);

        // A must be able to generate another message to advertise ReadOnly
        let msg = a.sync().generate_sync_message(&mut sa);
        assert!(
            msg.is_some(),
            "should generate message after set_read_only even with prior in_flight"
        );

        // Verify the message advertises ReadOnly
        let msg = msg.unwrap();
        assert!(msg.flags.unwrap().contains(MessageFlags::READ_ONLY));
    }

    #[test]
    fn generate_message_after_set_read_only_false_even_with_in_flight() {
        // Same as above but switching from read-only to read-write.
        let mut a = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut b = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        a.put(crate::ROOT, "from_a", "hello").unwrap();
        a.commit();

        let mut sa = State::new_read_only();
        let mut sb = State::new();

        // Sync while read-only
        sync(&mut a, &mut b, &mut sa, &mut sb);

        // B makes a change
        b.put(crate::ROOT, "from_b", "world").unwrap();
        b.commit();

        // A generates (sets in_flight)
        let _msg = a.sync().generate_sync_message(&mut sa);
        // Force some state by receiving a message
        let msg_b = b.sync().generate_sync_message(&mut sb).unwrap();
        a.sync().receive_sync_message(&mut sa, msg_b).unwrap();
        let _ = a.sync().generate_sync_message(&mut sa);

        // Now switch to read-write
        sa.set_read_only(false);

        // Must generate a message with SyncReset
        let msg = a.sync().generate_sync_message(&mut sa);
        assert!(
            msg.is_some(),
            "should generate message after switching to read-write"
        );
        let msg = msg.unwrap();
        let flags = msg.flags.unwrap();
        assert!(flags.contains(MessageFlags::SYNC_RESET));
        assert!(!flags.contains(MessageFlags::READ_ONLY));
    }

    #[test]
    fn both_toggle_read_only_to_read_write_simultaneously() {
        // Both peers start read-only with their own changes, sync (exchanging
        // heads but not changes), then both switch to read-write simultaneously.
        // After syncing again, both should have each other's changes.
        let mut doc1 = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut doc2 = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        doc1.put(crate::ROOT, "from_doc1", "hello").unwrap();
        doc1.commit();
        doc2.put(crate::ROOT, "from_doc2", "world").unwrap();
        doc2.commit();

        let mut s1 = State::new_read_only();
        let mut s2 = State::new_read_only();

        // Sync while both read-only — neither gets the other's changes
        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);
        assert!(doc1.get(crate::ROOT, "from_doc2").unwrap().is_none());
        assert!(doc2.get(crate::ROOT, "from_doc1").unwrap().is_none());

        // Both switch to read-write simultaneously
        s1.set_read_only(false);
        s2.set_read_only(false);

        // Sync again — both should exchange and apply changes
        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        assert!(doc1.get(crate::ROOT, "from_doc2").unwrap().is_some());
        assert!(doc2.get(crate::ROOT, "from_doc1").unwrap().is_some());
        assert_eq!(doc1.get_heads(), doc2.get_heads());
    }

    #[test]
    fn both_toggle_read_only_to_read_write_with_new_changes() {
        // Both peers start read-only, sync, then both switch to read-write
        // and make additional changes before syncing.
        let mut doc1 = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut doc2 = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        doc1.put(crate::ROOT, "original_1", "v1").unwrap();
        doc1.commit();
        doc2.put(crate::ROOT, "original_2", "v2").unwrap();
        doc2.commit();

        let mut s1 = State::new_read_only();
        let mut s2 = State::new_read_only();

        // Sync while both read-only
        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        // Both switch to read-write
        s1.set_read_only(false);
        s2.set_read_only(false);

        // Both make NEW changes after switching
        doc1.put(crate::ROOT, "new_1", "after_switch").unwrap();
        doc1.commit();
        doc2.put(crate::ROOT, "new_2", "after_switch").unwrap();
        doc2.commit();

        // Sync — should get both original AND new changes from each other
        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        assert!(doc1.get(crate::ROOT, "original_2").unwrap().is_some());
        assert!(doc1.get(crate::ROOT, "new_2").unwrap().is_some());
        assert!(doc2.get(crate::ROOT, "original_1").unwrap().is_some());
        assert!(doc2.get(crate::ROOT, "new_1").unwrap().is_some());
        assert_eq!(doc1.get_heads(), doc2.get_heads());
    }

    #[test]
    fn both_toggle_after_multiple_read_only_rounds() {
        // Both peers are read-only for several rounds, making changes each
        // round. Then both switch to read-write. All accumulated changes
        // from all rounds should be exchanged.
        let mut doc1 = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut doc2 = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        let mut s1 = State::new_read_only();
        let mut s2 = State::new_read_only();

        // Several rounds of read-only sync with changes each round
        for i in 0..5 {
            doc1.put(crate::ROOT, format!("doc1_r{i}"), i as i64)
                .unwrap();
            doc1.commit();
            doc2.put(crate::ROOT, format!("doc2_r{i}"), i as i64)
                .unwrap();
            doc2.commit();
            sync(&mut doc1, &mut doc2, &mut s1, &mut s2);
        }

        // Neither has the other's changes from any round
        for i in 0..5 {
            assert!(doc1
                .get(crate::ROOT, format!("doc2_r{i}"))
                .unwrap()
                .is_none());
            assert!(doc2
                .get(crate::ROOT, format!("doc1_r{i}"))
                .unwrap()
                .is_none());
        }

        // Both switch to read-write
        s1.set_read_only(false);
        s2.set_read_only(false);

        // Sync — all accumulated changes should flow
        sync(&mut doc1, &mut doc2, &mut s1, &mut s2);

        for i in 0..5 {
            assert!(
                doc1.get(crate::ROOT, format!("doc2_r{i}"))
                    .unwrap()
                    .is_some(),
                "doc1 missing doc2_r{i}"
            );
            assert!(
                doc2.get(crate::ROOT, format!("doc1_r{i}"))
                    .unwrap()
                    .is_some(),
                "doc2 missing doc1_r{i}"
            );
        }
        assert_eq!(doc1.get_heads(), doc2.get_heads());
    }

    #[test]
    fn switch_to_read_write_with_old_peer() {
        // Simulates switching from read-only to read-write when the remote
        // peer is an old implementation that doesn't understand SyncReset.
        // Old peers have their_capabilities = None. The fallback sends empty
        // heads, which triggers the old "peer lost all data" reset path.
        let mut a = crate::AutoCommit::new().with_actor(ActorId::try_from("abc123").unwrap());
        let mut b = crate::AutoCommit::new().with_actor(ActorId::try_from("def456").unwrap());

        a.put(crate::ROOT, "from_a", "hello").unwrap();
        a.commit();
        b.put(crate::ROOT, "from_b", "world").unwrap();
        b.commit();

        let mut sa = State::new_read_only();
        let mut sb = State::new();

        // Sync while read-only
        sync(&mut a, &mut b, &mut sa, &mut sb);
        assert!(a.get(crate::ROOT, "from_b").unwrap().is_none());
        assert!(b.get(crate::ROOT, "from_a").unwrap().is_some());

        // Simulate an old peer by clearing their_capabilities before switching.
        // set_read_only preserves their_capabilities, so we clear after.
        sa.set_read_only(false);
        sa.their_capabilities = None;

        // The first message should have empty heads (old peer fallback)
        let msg = a.sync().generate_sync_message(&mut sa).unwrap();
        assert!(msg.heads.is_empty(), "should send empty heads for old peer");
        assert!(
            !msg.flags.unwrap().contains(MessageFlags::SYNC_RESET),
            "should not include SyncReset for old peer"
        );

        // Complete the sync — old peer sees empty heads, clears sent_hashes
        b.sync().receive_sync_message(&mut sb, msg).unwrap();
        sync(&mut a, &mut b, &mut sa, &mut sb);

        // A should now have B's changes
        assert!(a.get(crate::ROOT, "from_b").unwrap().is_some());
        assert_eq!(a.get_heads(), b.get_heads());
    }
}