klukai_types/
broadcast.rs

1use std::{
2    cmp, fmt, io,
3    num::NonZeroU32,
4    num::ParseIntError,
5    ops::{Deref, RangeInclusive},
6    time::Duration,
7};
8
9use crate::api::{ColumnName, SqliteValue};
10use antithesis_sdk::assert_sometimes;
11use bytes::{Bytes, BytesMut};
12use foca::{Identity, Member, Notification, OwnedNotification, Runtime, Timer};
13use itertools::Itertools;
14use metrics::counter;
15use rusqlite::{
16    ToSql,
17    types::{FromSql, FromSqlError},
18};
19use serde::{Deserialize, Serialize};
20use speedy::{Context, Readable, Reader, Writable, Writer};
21use time::OffsetDateTime;
22use tokio::{
23    sync::{mpsc, oneshot},
24    task::block_in_place,
25};
26use tracing::{debug, error, trace};
27use uhlc::NTP64;
28
29use crate::{
30    actor::{Actor, ActorId, ClusterId},
31    agent::Agent,
32    base::{CrsqlDbVersion, CrsqlSeq},
33    change::{Change, ChunkedChanges, MAX_CHANGES_BYTE_SIZE, row_to_change},
34    channel::CorroSender,
35    sqlite::SqlitePoolError,
36    sync::SyncTraceContextV1,
37    updates::match_changes,
38};
39
40#[derive(Debug, Clone, Readable, Writable)]
41pub enum UniPayload {
42    V1 {
43        data: UniPayloadV1,
44        #[speedy(default_on_eof)]
45        cluster_id: ClusterId,
46    },
47}
48
49#[derive(Debug, Clone, Readable, Writable)]
50pub enum UniPayloadV1 {
51    Broadcast(BroadcastV1),
52}
53
54#[derive(Debug, Clone, Readable, Writable)]
55pub enum BiPayload {
56    V1 {
57        data: BiPayloadV1,
58        #[speedy(default_on_eof)]
59        cluster_id: ClusterId,
60    },
61}
62
63#[derive(Debug, Clone, Readable, Writable)]
64pub enum BiPayloadV1 {
65    SyncStart {
66        actor_id: ActorId,
67        #[speedy(default_on_eof)]
68        trace_ctx: SyncTraceContextV1,
69    },
70}
71
72#[derive(Debug)]
73pub enum FocaInput {
74    Announce(Actor),
75    Data(Bytes),
76    ClusterSize(NonZeroU32),
77    ApplyMany(Vec<Member<Actor>>),
78    Cmd(FocaCmd),
79}
80
81#[derive(Debug)]
82pub enum FocaCmd {
83    Rejoin(oneshot::Sender<Result<(), foca::Error>>),
84    MembershipStates(mpsc::Sender<foca::Member<Actor>>),
85    ChangeIdentity(Actor, oneshot::Sender<Result<(), foca::Error>>),
86}
87
88#[derive(Debug, Clone, Readable, Writable)]
89pub enum AuthzV1 {
90    Token(String),
91}
92
93#[derive(Clone, Debug, Readable, Writable)]
94pub enum BroadcastV1 {
95    Change(ChangeV1),
96}
97
98#[derive(Debug, Clone, PartialEq, Readable, Writable)]
99pub struct ColumnChange {
100    pub cid: ColumnName,
101    pub val: SqliteValue,
102    pub col_version: i64,
103    pub seq: CrsqlSeq,
104    pub cl: i64,
105}
106
107#[derive(Debug, Clone, Copy, strum::IntoStaticStr)]
108#[strum(serialize_all = "snake_case")]
109pub enum ChangeSource {
110    Broadcast,
111    Sync,
112}
113
114#[derive(Debug, Clone, PartialEq, Readable, Writable)]
115pub struct ChangeV1 {
116    pub actor_id: ActorId,
117    pub changeset: Changeset,
118}
119
120impl Deref for ChangeV1 {
121    type Target = Changeset;
122
123    fn deref(&self) -> &Self::Target {
124        &self.changeset
125    }
126}
127
128#[derive(Debug, Clone, PartialEq)]
129pub enum Changeset {
130    Empty {
131        versions: RangeInclusive<CrsqlDbVersion>,
132        ts: Option<Timestamp>,
133    },
134    Full {
135        version: CrsqlDbVersion,
136        changes: Vec<Change>,
137        // cr-sqlite sequences contained in this changeset
138        seqs: RangeInclusive<CrsqlSeq>,
139        // last cr-sqlite sequence for the complete changeset
140        last_seq: CrsqlSeq,
141        ts: Timestamp,
142    },
143    EmptySet {
144        versions: Vec<RangeInclusive<CrsqlDbVersion>>,
145        ts: Timestamp,
146    },
147}
148
149impl From<ChangesetParts> for Changeset {
150    fn from(value: ChangesetParts) -> Self {
151        Changeset::Full {
152            version: value.version,
153            changes: value.changes,
154            seqs: value.seqs,
155            last_seq: value.last_seq,
156            ts: value.ts,
157        }
158    }
159}
160
161pub struct ChangesetParts {
162    pub version: CrsqlDbVersion,
163    pub changes: Vec<Change>,
164    pub seqs: RangeInclusive<CrsqlSeq>,
165    pub last_seq: CrsqlSeq,
166    pub ts: Timestamp,
167}
168
169impl Changeset {
170    pub fn versions(&self) -> RangeInclusive<CrsqlDbVersion> {
171        match self {
172            Changeset::Empty { versions, .. } => versions.clone(),
173            // todo: this returns dummy version because empty set has an array of versions.
174            // probably shouldn't be doing this
175            Changeset::EmptySet { .. } => CrsqlDbVersion(0)..=CrsqlDbVersion(0),
176            Changeset::Full { version, .. } => *version..=*version,
177        }
178    }
179
180    // determine the estimated resource cost of processing a change
181    pub fn processing_cost(&self) -> usize {
182        match self {
183            Changeset::Empty { versions, .. } => {
184                cmp::min((versions.end().0 - versions.start().0) as usize + 1, 20)
185            }
186            Changeset::EmptySet { versions, .. } => versions
187                .iter()
188                .map(|versions| cmp::min((versions.end().0 - versions.start().0) as usize + 1, 20))
189                .sum::<usize>(),
190            Changeset::Full { changes, .. } => changes.len(),
191        }
192    }
193
194    pub fn max_db_version(&self) -> Option<CrsqlDbVersion> {
195        self.changes().iter().map(|c| c.db_version).max()
196    }
197
198    pub fn seqs(&self) -> Option<&RangeInclusive<CrsqlSeq>> {
199        match self {
200            Changeset::Empty { .. } => None,
201            Changeset::EmptySet { .. } => None,
202            Changeset::Full { seqs, .. } => Some(seqs),
203        }
204    }
205
206    pub fn last_seq(&self) -> Option<CrsqlSeq> {
207        match self {
208            Changeset::Empty { .. } => None,
209            Changeset::EmptySet { .. } => None,
210            Changeset::Full { last_seq, .. } => Some(*last_seq),
211        }
212    }
213
214    pub fn is_complete(&self) -> bool {
215        match self {
216            Changeset::Empty { .. } => true,
217            Changeset::EmptySet { .. } => true,
218            Changeset::Full { seqs, last_seq, .. } => {
219                *seqs.start() == CrsqlSeq(0) && seqs.end() == last_seq
220            }
221        }
222    }
223
224    pub fn len(&self) -> usize {
225        match self {
226            Changeset::Empty { .. } => 0, //(versions.end().0 - versions.start().0 + 1) as usize,
227            Changeset::EmptySet { versions, .. } => versions.len(),
228            Changeset::Full { changes, .. } => changes.len(),
229        }
230    }
231
232    pub fn is_empty(&self) -> bool {
233        match self {
234            Changeset::Empty { .. } => true,
235            Changeset::EmptySet { .. } => true,
236            Changeset::Full { changes, .. } => changes.is_empty(),
237        }
238    }
239
240    pub fn is_empty_set(&self) -> bool {
241        match self {
242            Changeset::Empty { .. } => false,
243            Changeset::EmptySet { .. } => true,
244            Changeset::Full { .. } => false,
245        }
246    }
247
248    pub fn ts(&self) -> Option<Timestamp> {
249        match self {
250            Changeset::Empty { ts, .. } => *ts,
251            Changeset::EmptySet { ts, .. } => Some(*ts),
252            Changeset::Full { ts, .. } => Some(*ts),
253        }
254    }
255
256    pub fn changes(&self) -> &[Change] {
257        match self {
258            Changeset::Empty { .. } => &[],
259            Changeset::EmptySet { .. } => &[],
260            Changeset::Full { changes, .. } => changes,
261        }
262    }
263
264    pub fn into_parts(self) -> Option<ChangesetParts> {
265        match self {
266            Changeset::Empty { .. } => None,
267            Changeset::EmptySet { .. } => None,
268            Changeset::Full {
269                version,
270                changes,
271                seqs,
272                last_seq,
273                ts,
274            } => Some(ChangesetParts {
275                version,
276                changes,
277                seqs,
278                last_seq,
279                ts,
280            }),
281        }
282    }
283}
284
285impl<'a, C> Readable<'a, C> for Changeset
286where
287    C: Context,
288{
289    fn read_from<R: Reader<'a, C>>(reader: &mut R) -> Result<Self, C::Error> {
290        let variant_tag = u8::read_from(reader)?;
291        match variant_tag {
292            0 => {
293                let start = CrsqlDbVersion::read_from(reader)?;
294                let end = CrsqlDbVersion::read_from(reader)?;
295                let versions = start..=end;
296                let ts = Option::<Timestamp>::read_from(reader)?;
297                Ok(Changeset::Empty { versions, ts })
298            }
299            1 => {
300                let version = CrsqlDbVersion::read_from(reader)?;
301                let changes = Vec::<Change>::read_from(reader)?;
302                let start_seq = CrsqlSeq::read_from(reader)?;
303                let end_seq = CrsqlSeq::read_from(reader)?;
304                let seqs = start_seq..=end_seq;
305                let last_seq = CrsqlSeq::read_from(reader)?;
306                let ts = Timestamp::read_from(reader)?;
307                Ok(Changeset::Full {
308                    version,
309                    changes,
310                    seqs,
311                    last_seq,
312                    ts,
313                })
314            }
315            2 => {
316                let versions_len = usize::read_from(reader)?;
317                let mut versions = Vec::with_capacity(versions_len);
318                for _ in 0..versions_len {
319                    let start = CrsqlDbVersion::read_from(reader)?;
320                    let end = CrsqlDbVersion::read_from(reader)?;
321                    versions.push(start..=end);
322                }
323                let ts = Timestamp::read_from(reader)?;
324                Ok(Changeset::EmptySet { versions, ts })
325            }
326            _ => {
327                // Read and discard the invalid tag to avoid issues, then create a proper error
328                let _ = reader.read_u8()?;
329                // This is a bit of a hack but should work for speedy contexts
330                panic!("Invalid changeset variant tag: {}", variant_tag);
331            }
332        }
333    }
334}
335
336impl<C> Writable<C> for Changeset
337where
338    C: Context,
339{
340    fn write_to<T: ?Sized + Writer<C>>(&self, writer: &mut T) -> Result<(), C::Error> {
341        match self {
342            Changeset::Empty { versions, ts } => {
343                0u8.write_to(writer)?;
344                versions.start().write_to(writer)?;
345                versions.end().write_to(writer)?;
346                ts.write_to(writer)?;
347            }
348            Changeset::Full {
349                version,
350                changes,
351                seqs,
352                last_seq,
353                ts,
354            } => {
355                1u8.write_to(writer)?;
356                version.write_to(writer)?;
357                changes.write_to(writer)?;
358                seqs.start().write_to(writer)?;
359                seqs.end().write_to(writer)?;
360                last_seq.write_to(writer)?;
361                ts.write_to(writer)?;
362            }
363            Changeset::EmptySet { versions, ts } => {
364                2u8.write_to(writer)?;
365                versions.len().write_to(writer)?;
366                for range in versions {
367                    range.start().write_to(writer)?;
368                    range.end().write_to(writer)?;
369                }
370                ts.write_to(writer)?;
371            }
372        }
373        Ok(())
374    }
375}
376
377#[derive(Debug, thiserror::Error)]
378pub enum TimestampParseError {
379    #[error("could not parse timestamp: {0:?}")]
380    Parse(ParseIntError),
381}
382
383#[derive(Debug, Default, Clone, Copy, Serialize, Deserialize, Eq, PartialOrd, Ord)]
384#[serde(transparent)]
385pub struct Timestamp(pub NTP64);
386
387impl Timestamp {
388    pub fn to_time(&self) -> OffsetDateTime {
389        OffsetDateTime::from_unix_timestamp(self.0.as_secs() as i64).unwrap()
390            + time::Duration::nanoseconds(self.0.subsec_nanos() as i64)
391    }
392
393    pub fn to_ntp64(&self) -> NTP64 {
394        self.0
395    }
396
397    pub fn zero() -> Self {
398        Timestamp::from(0u64)
399    }
400
401    pub fn is_zero(&self) -> bool {
402        self.0.0 == 0
403    }
404}
405
406// formatting to humantime and then parsing again incurs oddness, so lets compare secs and subsec_nanos
407impl PartialEq for Timestamp {
408    fn eq(&self, other: &Self) -> bool {
409        self.0.as_secs() == other.0.as_secs() && self.0.subsec_nanos() == other.0.subsec_nanos()
410    }
411}
412
413impl std::hash::Hash for Timestamp {
414    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
415        self.0.as_secs().hash(state);
416        self.0.subsec_nanos().hash(state);
417    }
418}
419
420impl Deref for Timestamp {
421    type Target = NTP64;
422
423    fn deref(&self) -> &Self::Target {
424        &self.0
425    }
426}
427
428impl fmt::Display for Timestamp {
429    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
430        self.0.fmt(f)
431    }
432}
433
434impl From<uhlc::Timestamp> for Timestamp {
435    fn from(ts: uhlc::Timestamp) -> Self {
436        Self(*ts.get_time())
437    }
438}
439
440impl From<NTP64> for Timestamp {
441    fn from(ntp64: NTP64) -> Self {
442        Self(ntp64)
443    }
444}
445
446impl From<u64> for Timestamp {
447    fn from(value: u64) -> Self {
448        Self(NTP64(value))
449    }
450}
451
452impl FromSql for Timestamp {
453    fn column_result(value: rusqlite::types::ValueRef<'_>) -> rusqlite::types::FromSqlResult<Self> {
454        match value {
455            rusqlite::types::ValueRef::Text(b) => match std::str::from_utf8(b) {
456                Ok(s) => match s.parse::<u64>() {
457                    Ok(ntp) => Ok(Timestamp(NTP64(ntp))),
458                    Err(e) => Err(FromSqlError::Other(Box::new(TimestampParseError::Parse(e)))),
459                },
460                Err(e) => Err(FromSqlError::Other(Box::new(e))),
461            },
462            _ => Err(FromSqlError::InvalidType),
463        }
464    }
465}
466
467impl ToSql for Timestamp {
468    fn to_sql(&self) -> rusqlite::Result<rusqlite::types::ToSqlOutput<'_>> {
469        Ok(rusqlite::types::ToSqlOutput::Owned(
470            rusqlite::types::Value::Text(self.0.as_u64().to_string()),
471        ))
472    }
473}
474
475impl<'a, C> Readable<'a, C> for Timestamp
476where
477    C: Context,
478{
479    #[inline]
480    fn read_from<R: Reader<'a, C>>(reader: &mut R) -> Result<Self, C::Error> {
481        Ok(Timestamp(NTP64(u64::read_from(reader)?)))
482    }
483
484    #[inline]
485    fn minimum_bytes_needed() -> usize {
486        std::mem::size_of::<u64>()
487    }
488}
489
490impl<C> Writable<C> for Timestamp
491where
492    C: Context,
493{
494    #[inline]
495    fn write_to<T: ?Sized + Writer<C>>(&self, writer: &mut T) -> Result<(), C::Error> {
496        self.0.0.write_to(writer)
497    }
498
499    #[inline]
500    fn bytes_needed(&self) -> Result<usize, C::Error> {
501        <u64 as speedy::Writable<C>>::bytes_needed(&self.0.0)
502    }
503}
504
505#[derive(Debug, thiserror::Error)]
506pub enum BroadcastEncodeError {
507    #[error(transparent)]
508    Encode(#[from] speedy::Error),
509    #[error(transparent)]
510    Io(#[from] io::Error),
511}
512
513#[derive(Debug, thiserror::Error)]
514pub enum BroadcastDecodeError {
515    #[error(transparent)]
516    Decode(#[from] speedy::Error),
517    #[error("corrupted message, crc mismatch (got: {0}, expected {1})")]
518    Corrupted(u32, u32),
519    #[error(transparent)]
520    Io(#[from] io::Error),
521    #[error("insufficient length received to decode message: {0}")]
522    InsufficientLength(usize),
523}
524
525#[derive(Debug)]
526pub enum BroadcastInput {
527    Rebroadcast(BroadcastV1),
528    AddBroadcast(BroadcastV1),
529}
530
531pub struct DispatchRuntime<T> {
532    pub to_send: CorroSender<(T, Bytes)>,
533    pub to_schedule: CorroSender<(Duration, Timer<T>)>,
534    pub notifications: CorroSender<OwnedNotification<T>>,
535    pub active: bool,
536    pub buf: BytesMut,
537}
538
539impl<T: Identity> Runtime<T> for DispatchRuntime<T> {
540    fn notify(&mut self, notification: Notification<'_, T>) {
541        match &notification {
542            Notification::Active => {
543                self.active = true;
544            }
545            Notification::Idle | Notification::Defunct => {
546                self.active = false;
547            }
548            _ => {}
549        };
550
551        // Convert the borrowed notification to an owned one for sending through the channel
552        let owned_notification = notification.to_owned();
553
554        if let Err(e) = self.notifications.try_send(owned_notification) {
555            counter!("corro.channel.error", "type" => "full", "name" => "dispatch.notifications")
556                .increment(1);
557            error!("error dispatching notification: {e}");
558        }
559    }
560
561    fn send_to(&mut self, to: T, data: &[u8]) {
562        trace!("cluster send_to {to:?}");
563        self.buf.extend_from_slice(data);
564
565        if let Err(e) = self.to_send.try_send((to, self.buf.split().freeze())) {
566            counter!("corro.channel.error", "type" => "full", "name" => "dispatch.to_send")
567                .increment(1);
568            error!("error dispatching broadcast packet: {e}");
569        }
570    }
571
572    fn submit_after(&mut self, event: Timer<T>, after: Duration) {
573        if let Err(e) = self.to_schedule.try_send((after, event)) {
574            counter!("corro.channel.error", "type" => "full", "name" => "dispatch.to_schedule")
575                .increment(1);
576            error!("error dispatching scheduled event: {e}");
577        }
578    }
579}
580
581impl<T> DispatchRuntime<T> {
582    pub fn new(
583        to_send: CorroSender<(T, Bytes)>,
584        to_schedule: CorroSender<(Duration, Timer<T>)>,
585        notifications: CorroSender<OwnedNotification<T>>,
586    ) -> Self {
587        Self {
588            to_send,
589            to_schedule,
590            notifications,
591            active: false,
592            buf: BytesMut::new(),
593        }
594    }
595}
596
597#[derive(Debug, thiserror::Error)]
598pub enum BroadcastError {
599    #[error(transparent)]
600    Pool(#[from] SqlitePoolError),
601    #[error(transparent)]
602    Sqlite(#[from] rusqlite::Error),
603}
604
605pub async fn broadcast_changes(
606    agent: Agent,
607    db_version: CrsqlDbVersion,
608    last_seq: CrsqlSeq,
609    ts: Timestamp,
610) -> Result<(), BroadcastError> {
611    let actor_id = agent.actor_id();
612    let conn = agent.pool().read().await?;
613    trace!("got conn for broadcast");
614
615    block_in_place(|| {
616        // TODO: make this more generic so both sync and local changes can use it.
617        let mut prepped = conn.prepare_cached(
618            r#"
619                SELECT "table", pk, cid, val, col_version, db_version, seq, site_id, cl
620                    FROM crsql_changes
621                    WHERE db_version = ?
622                    AND site_id = crsql_site_id()
623                    ORDER BY seq ASC
624            "#,
625        )?;
626        let rows = prepped.query_map([db_version], row_to_change)?;
627        let chunked = ChunkedChanges::new(rows, CrsqlSeq(0), last_seq, MAX_CHANGES_BYTE_SIZE);
628        for changes_seqs in chunked {
629            match changes_seqs {
630                Ok((changes, seqs)) => {
631                    for (table_name, count) in changes.iter().counts_by(|change| &change.table) {
632                        counter!("corro.changes.committed", "table" => table_name.to_string(), "source" => "local").increment(count as u64);
633                    }
634
635                    trace!("broadcasting changes: {changes:?} for seq: {seqs:?}");
636
637                    debug!("match_changes db_version: {db_version}");
638                    match_changes(agent.subs_manager(), &changes, db_version);
639                    match_changes(agent.updates_manager(), &changes, db_version);
640
641                    let tx_bcast = agent.tx_bcast().clone();
642                    assert_sometimes!(true, "Corrosion broadcasts changes");
643                    tokio::spawn(async move {
644                        if let Err(e) = tx_bcast
645                            .send(BroadcastInput::AddBroadcast(BroadcastV1::Change(
646                                ChangeV1 {
647                                    actor_id,
648                                    changeset: Changeset::Full {
649                                        version: db_version,
650                                        changes,
651                                        seqs,
652                                        last_seq,
653                                        ts,
654                                    },
655                                },
656                            )))
657                            .await
658                        {
659                            error!("could not send change message for broadcast: {e}");
660                        }
661                    });
662                }
663                Err(e) => {
664                    error!(
665                        "could not process crsql change (db_version: {db_version}) for broadcast: {e}"
666                    );
667                    break;
668                }
669            }
670        }
671        Ok::<_, rusqlite::Error>(())
672    })?;
673
674    Ok(())
675}
676
677#[cfg(test)]
678mod tests {
679    use super::*;
680    use speedy::{Readable, Writable};
681
682    #[test]
683    fn test_changeset_empty_serialization() {
684        let changeset = Changeset::Empty {
685            versions: CrsqlDbVersion(10)..=CrsqlDbVersion(20),
686            ts: Some(Timestamp::from(12345u64)),
687        };
688
689        // Serialize
690        let bytes = changeset.write_to_vec().unwrap();
691
692        // Deserialize
693        let deserialized = Changeset::read_from_buffer(&bytes).unwrap();
694
695        assert_eq!(changeset, deserialized);
696    }
697
698    #[test]
699    fn test_changeset_empty_with_none_ts() {
700        let changeset = Changeset::Empty {
701            versions: CrsqlDbVersion(5)..=CrsqlDbVersion(5),
702            ts: None,
703        };
704
705        let bytes = changeset.write_to_vec().unwrap();
706        let deserialized = Changeset::read_from_buffer(&bytes).unwrap();
707
708        assert_eq!(changeset, deserialized);
709    }
710
711    #[test]
712    fn test_changeset_full_serialization() {
713        let changeset = Changeset::Full {
714            version: CrsqlDbVersion(42),
715            changes: vec![], // Empty changes for simplicity
716            seqs: CrsqlSeq(100)..=CrsqlSeq(150),
717            last_seq: CrsqlSeq(200),
718            ts: Timestamp::from(67890u64),
719        };
720
721        let bytes = changeset.write_to_vec().unwrap();
722        let deserialized = Changeset::read_from_buffer(&bytes).unwrap();
723
724        assert_eq!(changeset, deserialized);
725    }
726
727    #[test]
728    fn test_changeset_empty_set_serialization() {
729        let changeset = Changeset::EmptySet {
730            versions: vec![
731                CrsqlDbVersion(1)..=CrsqlDbVersion(5),
732                CrsqlDbVersion(10)..=CrsqlDbVersion(15),
733                CrsqlDbVersion(20)..=CrsqlDbVersion(20),
734            ],
735            ts: Timestamp::from(11111u64),
736        };
737
738        let bytes = changeset.write_to_vec().unwrap();
739        let deserialized = Changeset::read_from_buffer(&bytes).unwrap();
740
741        assert_eq!(changeset, deserialized);
742    }
743
744    #[test]
745    fn test_changeset_empty_set_with_empty_vec() {
746        let changeset = Changeset::EmptySet {
747            versions: vec![],
748            ts: Timestamp::from(22222u64),
749        };
750
751        let bytes = changeset.write_to_vec().unwrap();
752        let deserialized = Changeset::read_from_buffer(&bytes).unwrap();
753
754        assert_eq!(changeset, deserialized);
755    }
756
757    #[test]
758    fn test_changeset_roundtrip_all_variants() {
759        let test_cases = vec![
760            Changeset::Empty {
761                versions: CrsqlDbVersion(1)..=CrsqlDbVersion(1),
762                ts: None,
763            },
764            Changeset::Full {
765                version: CrsqlDbVersion(1),
766                changes: vec![],
767                seqs: CrsqlSeq(0)..=CrsqlSeq(0),
768                last_seq: CrsqlSeq(0),
769                ts: Timestamp::zero(),
770            },
771            Changeset::EmptySet {
772                versions: vec![CrsqlDbVersion(1)..=CrsqlDbVersion(3)],
773                ts: Timestamp::from(99999u64),
774            },
775        ];
776
777        for (i, original) in test_cases.into_iter().enumerate() {
778            let bytes = original.write_to_vec().unwrap();
779            let deserialized = Changeset::read_from_buffer(&bytes)
780                .unwrap_or_else(|e| panic!("Failed to deserialize test case {}: {}", i, e));
781
782            assert_eq!(original, deserialized, "Test case {} failed", i);
783        }
784    }
785}
klukai_types/broadcast.rs

klukai_types/
broadcast.rs
