klukai_types/
broadcast.rs

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

klukai_types/
broadcast.rs
