klukai_types/
broadcast.rs

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

klukai_types/
broadcast.rs
