zakura_network/zakura/
exchange.rs

1//! Shared Zakura sync frontier contract.
2
3use std::sync::{
4    atomic::{AtomicU64, Ordering},
5    Arc,
6};
7
8use futures::future::BoxFuture;
9use serde_json::{Number, Value};
10use tokio::sync::watch;
11use zakura_chain::block;
12
13use super::{
14    commit_state_trace as cs_trace, BlockApplyResult, BlockApplyToken, BlockSyncBlockMeta,
15    HeaderSyncCommitFailureKind, ZakuraPeerId, ZakuraTrace, COMMIT_STATE_TABLE,
16};
17
18/// A height/hash pair at one chain frontier.
19#[derive(Copy, Clone, Debug, Eq, PartialEq)]
20pub struct Frontier {
21    /// Frontier height.
22    pub height: block::Height,
23    /// Frontier block hash.
24    pub hash: block::Hash,
25}
26
27impl Frontier {
28    /// Construct a frontier from its height and hash.
29    pub fn new(height: block::Height, hash: block::Hash) -> Self {
30        Self { height, hash }
31    }
32}
33
34/// Shared Zakura chain facts owned by the sync exchange.
35#[derive(Copy, Clone, Debug, Eq, PartialEq)]
36pub struct ChainFrontier {
37    /// Highest finalized block.
38    pub finalized: Frontier,
39    /// Highest verified block body.
40    pub verified_body: Frontier,
41    /// Highest committed header target.
42    pub best_header: Frontier,
43}
44
45/// Cause for a shared frontier update.
46#[derive(Copy, Clone, Debug, Eq, PartialEq)]
47pub enum FrontierChange {
48    /// Initial or whole-state snapshot.
49    Snapshot,
50    /// Verified body frontier advanced.
51    VerifiedGrow,
52    /// Verified body frontier was reset, possibly lower.
53    VerifiedReset,
54    /// Best header target advanced.
55    HeaderAdvanced,
56    /// Best header target was reanchored, possibly lower.
57    HeaderReanchored,
58}
59
60/// Latest shared frontier plus the transition cause.
61#[derive(Copy, Clone, Debug, Eq, PartialEq)]
62pub struct FrontierUpdate {
63    /// Current shared frontier after the change.
64    pub frontier: ChainFrontier,
65    /// Cause of the current value.
66    pub change: FrontierChange,
67}
68
69/// Single owner of shared Zakura sync frontiers.
70#[derive(Clone, Debug)]
71pub struct ZakuraSyncExchange {
72    inner: Arc<ZakuraSyncExchangeInner>,
73}
74
75#[derive(Debug)]
76struct ZakuraSyncExchangeInner {
77    frontier: watch::Sender<FrontierUpdate>,
78    trace: ZakuraTrace,
79    sequence: AtomicU64,
80}
81
82impl ZakuraSyncExchange {
83    /// Create a shared sync exchange with an initial coherent frontier snapshot.
84    pub fn new(initial: FrontierUpdate, trace: ZakuraTrace) -> Self {
85        let (frontier, _receiver) = watch::channel(initial);
86
87        Self {
88            inner: Arc::new(ZakuraSyncExchangeInner {
89                frontier,
90                trace,
91                sequence: AtomicU64::new(0),
92            }),
93        }
94    }
95
96    /// Return the currently cached shared frontier update.
97    pub fn current_frontier(&self) -> FrontierUpdate {
98        *self.inner.frontier.borrow()
99    }
100
101    /// Subscribe to latest-value frontier updates.
102    pub fn subscribe_frontier(&self) -> watch::Receiver<FrontierUpdate> {
103        self.inner.frontier.subscribe()
104    }
105
106    /// Publish a candidate frontier update from `source`.
107    pub fn publish_frontier(&self, requested: FrontierUpdate, source: &'static str) {
108        let mut transition = None;
109
110        self.inner.frontier.send_if_modified(|current| {
111            let old = *current;
112            let sequence = self
113                .inner
114                .sequence
115                .fetch_add(1, Ordering::Relaxed)
116                .saturating_add(1);
117
118            match apply_frontier_update(old, requested) {
119                Some(update) => {
120                    *current = update;
121                    transition = Some((sequence, old, update, "accepted"));
122                    true
123                }
124                None => {
125                    let ignored = FrontierUpdate {
126                        frontier: old.frontier,
127                        change: requested.change,
128                    };
129                    transition = Some((sequence, old, ignored, "ignored"));
130                    false
131                }
132            }
133        });
134
135        if let Some((sequence, old, new, result)) = transition {
136            self.trace_transition(sequence, source, old, new, result);
137        }
138    }
139
140    fn trace_transition(
141        &self,
142        sequence: u64,
143        source: &'static str,
144        old: FrontierUpdate,
145        new: FrontierUpdate,
146        result: &'static str,
147    ) {
148        self.inner.trace.emit_with(COMMIT_STATE_TABLE, |row| {
149            row.insert(
150                cs_trace::EVENT.to_string(),
151                Value::String(cs_trace::SYNC_FRONTIER_TRANSITION.to_string()),
152            );
153            row.insert(
154                cs_trace::SOURCE.to_string(),
155                Value::String(source.to_string()),
156            );
157            row.insert(
158                cs_trace::SEQUENCE.to_string(),
159                Value::Number(Number::from(sequence)),
160            );
161            row.insert(
162                cs_trace::CAUSE.to_string(),
163                Value::String(frontier_change_label(new.change).to_string()),
164            );
165            row.insert(
166                cs_trace::RESULT.to_string(),
167                Value::String(result.to_string()),
168            );
169            insert_frontier_fields(row, "old", old.frontier);
170            insert_frontier_fields(row, "new", new.frontier);
171        });
172    }
173}
174
175/// Result of a header range commit through the sync exchange.
176#[derive(Copy, Clone, Debug, Eq, PartialEq)]
177pub struct HeaderRangeCommit {
178    /// First committed header height.
179    pub start_height: block::Height,
180    /// New durable best header frontier.
181    pub tip: Frontier,
182}
183
184/// Result of a submitted block body through the sync exchange.
185#[derive(Copy, Clone, Debug, Eq, PartialEq)]
186pub struct BlockBodySubmit {
187    /// Submission token echoed from the reactor.
188    pub token: BlockApplyToken,
189    /// Submitted block height.
190    pub height: block::Height,
191    /// Submitted block hash.
192    pub hash: block::Hash,
193    /// Verifier result.
194    pub result: BlockApplyResult,
195    /// Locally observed shared frontier after the apply attempt.
196    pub local_frontier: Option<ChainFrontier>,
197}
198
199/// Header-sync view of shared Zakura state.
200pub trait HeaderSyncStatePortImpl: Send + Sync + 'static {
201    /// Return the currently cached shared frontier.
202    fn current_frontier(&self) -> ChainFrontier;
203
204    /// Subscribe to latest-value shared frontier updates.
205    fn subscribe_frontier(&self) -> watch::Receiver<FrontierUpdate>;
206
207    /// Commit a contiguous header range.
208    fn commit_header_range(
209        &self,
210        peer: ZakuraPeerId,
211        anchor: block::Hash,
212        start_height: block::Height,
213        headers: Vec<Arc<block::Header>>,
214        body_sizes: Vec<u32>,
215        finalized: bool,
216    ) -> BoxFuture<'static, Result<HeaderRangeCommit, HeaderSyncCommitFailureKind>>;
217
218    /// Publish a locally accepted best-header advance.
219    fn publish_best_header(&self, tip: Frontier) -> BoxFuture<'static, ()>;
220
221    /// Publish a best-header reanchor.
222    fn publish_header_reanchor(&self, old: Frontier, new: Frontier) -> BoxFuture<'static, ()>;
223}
224
225/// Block-sync view of shared Zakura state.
226pub trait BlockSyncStatePortImpl: Send + Sync + 'static {
227    /// Return the currently cached shared frontier.
228    fn current_frontier(&self) -> ChainFrontier;
229
230    /// Subscribe to latest-value shared frontier updates.
231    fn subscribe_frontier(&self) -> watch::Receiver<FrontierUpdate>;
232
233    /// Query committed headers that still need block bodies.
234    fn query_missing_bodies(
235        &self,
236        verified_body: block::Height,
237        best_header: block::Height,
238    ) -> BoxFuture<'static, Result<Vec<BlockSyncBlockMeta>, zakura_chain::BoxError>>;
239
240    /// Submit a downloaded body to the verifier/state pipeline.
241    fn submit_block_body(
242        &self,
243        token: BlockApplyToken,
244        block: Arc<block::Block>,
245    ) -> BoxFuture<'static, BlockBodySubmit>;
246
247    /// Read committed blocks for serving stream-6 peers.
248    fn read_committed_blocks(
249        &self,
250        start: block::Height,
251        count: u32,
252    ) -> BoxFuture<
253        'static,
254        Result<Vec<(block::Height, Arc<block::Block>, usize)>, zakura_chain::BoxError>,
255    >;
256
257    /// Publish body-sync progress that did not come from a direct submit path.
258    fn publish_body_progress(&self, update: FrontierUpdate) -> BoxFuture<'static, ()>;
259}
260
261/// Cloneable header-sync state port.
262#[derive(Clone)]
263pub struct HeaderSyncStatePort {
264    inner: Arc<dyn HeaderSyncStatePortImpl>,
265}
266
267impl HeaderSyncStatePort {
268    /// Wrap an implementation in a cloneable port.
269    pub fn new(inner: Arc<dyn HeaderSyncStatePortImpl>) -> Self {
270        Self { inner }
271    }
272
273    /// Return the currently cached shared frontier.
274    pub fn current_frontier(&self) -> ChainFrontier {
275        self.inner.current_frontier()
276    }
277
278    /// Subscribe to latest-value shared frontier updates.
279    pub fn subscribe_frontier(&self) -> watch::Receiver<FrontierUpdate> {
280        self.inner.subscribe_frontier()
281    }
282
283    /// Commit a contiguous header range.
284    pub fn commit_header_range(
285        &self,
286        peer: ZakuraPeerId,
287        anchor: block::Hash,
288        start_height: block::Height,
289        headers: Vec<Arc<block::Header>>,
290        body_sizes: Vec<u32>,
291        finalized: bool,
292    ) -> BoxFuture<'static, Result<HeaderRangeCommit, HeaderSyncCommitFailureKind>> {
293        self.inner
294            .commit_header_range(peer, anchor, start_height, headers, body_sizes, finalized)
295    }
296
297    /// Publish a locally accepted best-header advance.
298    pub fn publish_best_header(&self, tip: Frontier) -> BoxFuture<'static, ()> {
299        self.inner.publish_best_header(tip)
300    }
301
302    /// Publish a best-header reanchor.
303    pub fn publish_header_reanchor(&self, old: Frontier, new: Frontier) -> BoxFuture<'static, ()> {
304        self.inner.publish_header_reanchor(old, new)
305    }
306}
307
308impl std::fmt::Debug for HeaderSyncStatePort {
309    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
310        formatter.write_str("HeaderSyncStatePort")
311    }
312}
313
314/// Cloneable block-sync state port.
315#[derive(Clone)]
316pub struct BlockSyncStatePort {
317    inner: Arc<dyn BlockSyncStatePortImpl>,
318}
319
320impl BlockSyncStatePort {
321    /// Wrap an implementation in a cloneable port.
322    pub fn new(inner: Arc<dyn BlockSyncStatePortImpl>) -> Self {
323        Self { inner }
324    }
325
326    /// Return the currently cached shared frontier.
327    pub fn current_frontier(&self) -> ChainFrontier {
328        self.inner.current_frontier()
329    }
330
331    /// Subscribe to latest-value shared frontier updates.
332    pub fn subscribe_frontier(&self) -> watch::Receiver<FrontierUpdate> {
333        self.inner.subscribe_frontier()
334    }
335
336    /// Query committed headers that still need block bodies.
337    pub fn query_missing_bodies(
338        &self,
339        verified_body: block::Height,
340        best_header: block::Height,
341    ) -> BoxFuture<'static, Result<Vec<BlockSyncBlockMeta>, zakura_chain::BoxError>> {
342        self.inner.query_missing_bodies(verified_body, best_header)
343    }
344
345    /// Submit a downloaded body to the verifier/state pipeline.
346    pub fn submit_block_body(
347        &self,
348        token: BlockApplyToken,
349        block: Arc<block::Block>,
350    ) -> BoxFuture<'static, BlockBodySubmit> {
351        self.inner.submit_block_body(token, block)
352    }
353
354    /// Read committed blocks for serving stream-6 peers.
355    pub fn read_committed_blocks(
356        &self,
357        start: block::Height,
358        count: u32,
359    ) -> BoxFuture<
360        'static,
361        Result<Vec<(block::Height, Arc<block::Block>, usize)>, zakura_chain::BoxError>,
362    > {
363        self.inner.read_committed_blocks(start, count)
364    }
365
366    /// Publish body-sync progress that did not come from a direct submit path.
367    pub fn publish_body_progress(&self, update: FrontierUpdate) -> BoxFuture<'static, ()> {
368        self.inner.publish_body_progress(update)
369    }
370}
371
372impl std::fmt::Debug for BlockSyncStatePort {
373    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
374        formatter.write_str("BlockSyncStatePort")
375    }
376}
377
378/// Build a [`ChainFrontier`] from startup pieces that do not expose finalized hashes.
379pub fn chain_frontier_from_parts(
380    finalized_height: block::Height,
381    verified_body: Frontier,
382    best_header: Frontier,
383) -> ChainFrontier {
384    let finalized_hash = if finalized_height == verified_body.height {
385        verified_body.hash
386    } else {
387        block::Hash([0; 32])
388    };
389
390    ChainFrontier {
391        finalized: Frontier::new(finalized_height, finalized_hash),
392        verified_body,
393        best_header,
394    }
395}
396
397/// Apply one requested frontier update to the current exchange frontier.
398///
399/// Returns `None` when the requested update is stale or redundant.
400pub fn apply_frontier_update(
401    current: FrontierUpdate,
402    requested: FrontierUpdate,
403) -> Option<FrontierUpdate> {
404    let mut frontier = current.frontier;
405
406    match requested.change {
407        FrontierChange::Snapshot => {
408            frontier.finalized =
409                higher_frontier(current.frontier.finalized, requested.frontier.finalized);
410            frontier.verified_body = higher_frontier(
411                current.frontier.verified_body,
412                requested.frontier.verified_body,
413            );
414            frontier.best_header =
415                higher_frontier(current.frontier.best_header, requested.frontier.best_header);
416            if frontier == current.frontier {
417                return None;
418            }
419        }
420        FrontierChange::VerifiedGrow => {
421            if requested.frontier.verified_body.height < current.frontier.verified_body.height {
422                return None;
423            }
424            frontier.finalized =
425                higher_frontier(current.frontier.finalized, requested.frontier.finalized);
426            frontier.verified_body = requested.frontier.verified_body;
427        }
428        FrontierChange::VerifiedReset => {
429            frontier.finalized =
430                higher_frontier(current.frontier.finalized, requested.frontier.finalized);
431            frontier.verified_body = requested.frontier.verified_body;
432        }
433        FrontierChange::HeaderAdvanced => {
434            if requested.frontier.best_header.height <= current.frontier.best_header.height {
435                return None;
436            }
437            frontier.best_header = requested.frontier.best_header;
438        }
439        FrontierChange::HeaderReanchored => {
440            if requested.frontier.best_header == current.frontier.best_header {
441                return None;
442            }
443            frontier.best_header = requested.frontier.best_header;
444        }
445    }
446
447    Some(FrontierUpdate {
448        frontier,
449        change: requested.change,
450    })
451}
452
453fn higher_frontier(left: Frontier, right: Frontier) -> Frontier {
454    if right.height > left.height {
455        right
456    } else {
457        left
458    }
459}
460
461fn frontier_change_label(change: FrontierChange) -> &'static str {
462    match change {
463        FrontierChange::Snapshot => "snapshot",
464        FrontierChange::VerifiedGrow => "verified_grow",
465        FrontierChange::VerifiedReset => "verified_reset",
466        FrontierChange::HeaderAdvanced => "header_advanced",
467        FrontierChange::HeaderReanchored => "header_reanchored",
468    }
469}
470
471fn insert_frontier_fields(
472    row: &mut serde_json::Map<String, Value>,
473    prefix: &'static str,
474    frontier: ChainFrontier,
475) {
476    let (
477        finalized_height,
478        finalized_hash,
479        verified_body_height,
480        verified_body_hash,
481        best_header_height,
482        best_header_hash,
483    ) = match prefix {
484        "old" => (
485            cs_trace::OLD_FINALIZED_HEIGHT,
486            cs_trace::OLD_FINALIZED_HASH,
487            cs_trace::OLD_VERIFIED_BODY_HEIGHT,
488            cs_trace::OLD_VERIFIED_BODY_HASH,
489            cs_trace::OLD_BEST_HEADER_HEIGHT,
490            cs_trace::OLD_BEST_HEADER_HASH,
491        ),
492        "new" => (
493            cs_trace::NEW_FINALIZED_HEIGHT,
494            cs_trace::NEW_FINALIZED_HASH,
495            cs_trace::NEW_VERIFIED_BODY_HEIGHT,
496            cs_trace::NEW_VERIFIED_BODY_HASH,
497            cs_trace::NEW_BEST_HEADER_HEIGHT,
498            cs_trace::NEW_BEST_HEADER_HASH,
499        ),
500        _ => return,
501    };
502
503    insert_height(row, finalized_height, frontier.finalized.height);
504    insert_hash(row, finalized_hash, frontier.finalized.hash);
505    insert_height(row, verified_body_height, frontier.verified_body.height);
506    insert_hash(row, verified_body_hash, frontier.verified_body.hash);
507    insert_height(row, best_header_height, frontier.best_header.height);
508    insert_hash(row, best_header_hash, frontier.best_header.hash);
509}
510
511fn insert_height(
512    row: &mut serde_json::Map<String, Value>,
513    key: &'static str,
514    height: block::Height,
515) {
516    row.insert(
517        key.to_string(),
518        Value::Number(Number::from(u64::from(height.0))),
519    );
520}
521
522fn insert_hash(row: &mut serde_json::Map<String, Value>, key: &'static str, hash: block::Hash) {
523    row.insert(key.to_string(), Value::String(format!("{hash}")));
524}
525
526#[cfg(test)]
527mod tests {
528    use super::*;
529    use std::{
530        sync::{Arc, Barrier},
531        thread,
532    };
533
534    fn frontier(height: u32, seed: u8) -> Frontier {
535        Frontier::new(block::Height(height), block::Hash([seed; 32]))
536    }
537
538    fn update(
539        finalized: Frontier,
540        verified_body: Frontier,
541        best_header: Frontier,
542        change: FrontierChange,
543    ) -> FrontierUpdate {
544        FrontierUpdate {
545            frontier: ChainFrontier {
546                finalized,
547                verified_body,
548                best_header,
549            },
550            change,
551        }
552    }
553
554    #[test]
555    fn grow_advances_verified_body() {
556        let current = update(
557            frontier(1, 1),
558            frontier(2, 2),
559            frontier(5, 5),
560            FrontierChange::Snapshot,
561        );
562        let requested = update(
563            frontier(3, 3),
564            frontier(4, 4),
565            frontier(9, 9),
566            FrontierChange::VerifiedGrow,
567        );
568
569        let updated = apply_frontier_update(current, requested).expect("grow is accepted");
570
571        assert_eq!(updated.frontier.verified_body, frontier(4, 4));
572        assert_eq!(updated.frontier.best_header, frontier(5, 5));
573    }
574
575    #[test]
576    fn grow_may_not_lower_best_header() {
577        let current = update(
578            frontier(1, 1),
579            frontier(2, 2),
580            frontier(9, 9),
581            FrontierChange::Snapshot,
582        );
583        let requested = update(
584            frontier(3, 3),
585            frontier(4, 4),
586            frontier(5, 5),
587            FrontierChange::VerifiedGrow,
588        );
589
590        let updated = apply_frontier_update(current, requested).expect("grow is accepted");
591
592        assert_eq!(updated.frontier.finalized, frontier(3, 3));
593        assert_eq!(updated.frontier.verified_body, frontier(4, 4));
594        assert_eq!(updated.frontier.best_header, frontier(9, 9));
595    }
596
597    #[test]
598    fn stale_lower_grow_is_ignored() {
599        let current = update(
600            frontier(1, 1),
601            frontier(8, 8),
602            frontier(9, 9),
603            FrontierChange::Snapshot,
604        );
605        let requested = update(
606            frontier(1, 1),
607            frontier(7, 7),
608            frontier(9, 9),
609            FrontierChange::VerifiedGrow,
610        );
611
612        assert!(apply_frontier_update(current, requested).is_none());
613    }
614
615    #[test]
616    fn equal_height_grow_refreshes_verified_hash() {
617        let current = update(
618            frontier(1, 1),
619            frontier(8, 8),
620            frontier(9, 9),
621            FrontierChange::Snapshot,
622        );
623        let requested = update(
624            frontier(1, 1),
625            frontier(8, 18),
626            frontier(9, 19),
627            FrontierChange::VerifiedGrow,
628        );
629
630        let updated =
631            apply_frontier_update(current, requested).expect("equal height grow is accepted");
632
633        assert_eq!(updated.frontier.verified_body, frontier(8, 18));
634        assert_eq!(updated.frontier.best_header, frontier(9, 9));
635    }
636
637    #[test]
638    fn reset_may_lower_verified_body() {
639        let current = update(
640            frontier(5, 5),
641            frontier(8, 8),
642            frontier(10, 10),
643            FrontierChange::Snapshot,
644        );
645        let requested = update(
646            frontier(4, 4),
647            frontier(6, 6),
648            frontier(2, 2),
649            FrontierChange::VerifiedReset,
650        );
651
652        let updated = apply_frontier_update(current, requested).expect("reset is accepted");
653
654        assert_eq!(updated.frontier.finalized, frontier(5, 5));
655        assert_eq!(updated.frontier.verified_body, frontier(6, 6));
656        assert_eq!(updated.frontier.best_header, frontier(10, 10));
657    }
658
659    #[test]
660    fn finalized_height_never_decreases_across_mixed_updates() {
661        let mut current = update(
662            frontier(10, 10),
663            frontier(10, 10),
664            frontier(12, 12),
665            FrontierChange::Snapshot,
666        );
667        for requested in [
668            update(
669                frontier(9, 9),
670                frontier(11, 11),
671                frontier(12, 12),
672                FrontierChange::VerifiedGrow,
673            ),
674            update(
675                frontier(1, 1),
676                frontier(7, 7),
677                frontier(0, 0),
678                FrontierChange::VerifiedReset,
679            ),
680            update(
681                frontier(8, 8),
682                frontier(8, 8),
683                frontier(20, 20),
684                FrontierChange::HeaderAdvanced,
685            ),
686            update(
687                frontier(3, 3),
688                frontier(3, 3),
689                frontier(13, 13),
690                FrontierChange::HeaderReanchored,
691            ),
692        ] {
693            if let Some(updated) = apply_frontier_update(current, requested) {
694                current = updated;
695                assert!(
696                    current.frontier.finalized.height >= block::Height(10),
697                    "finalized frontier must never move below the original finalized height"
698                );
699            }
700        }
701    }
702
703    #[test]
704    fn snapshot_does_not_lower_existing_frontiers() {
705        let current = update(
706            frontier(5, 5),
707            frontier(8, 8),
708            frontier(12, 12),
709            FrontierChange::Snapshot,
710        );
711        let requested = update(
712            frontier(4, 4),
713            frontier(7, 7),
714            frontier(11, 11),
715            FrontierChange::Snapshot,
716        );
717
718        assert!(apply_frontier_update(current, requested).is_none());
719    }
720
721    #[test]
722    fn header_reanchor_lowers_only_best_header() {
723        let current = update(
724            frontier(5, 5),
725            frontier(8, 8),
726            frontier(12, 12),
727            FrontierChange::Snapshot,
728        );
729        let requested = update(
730            frontier(1, 1),
731            frontier(2, 2),
732            frontier(9, 9),
733            FrontierChange::HeaderReanchored,
734        );
735
736        let updated = apply_frontier_update(current, requested).expect("reanchor is accepted");
737
738        assert_eq!(updated.frontier.finalized, frontier(5, 5));
739        assert_eq!(updated.frontier.verified_body, frontier(8, 8));
740        assert_eq!(updated.frontier.best_header, frontier(9, 9));
741    }
742
743    #[test]
744    fn header_advance_cannot_change_verified_body() {
745        let current = update(
746            frontier(5, 5),
747            frontier(8, 8),
748            frontier(12, 12),
749            FrontierChange::Snapshot,
750        );
751        let requested = update(
752            frontier(1, 1),
753            frontier(2, 2),
754            frontier(13, 13),
755            FrontierChange::HeaderAdvanced,
756        );
757
758        let updated = apply_frontier_update(current, requested).expect("advance is accepted");
759
760        assert_eq!(updated.frontier.finalized, frontier(5, 5));
761        assert_eq!(updated.frontier.verified_body, frontier(8, 8));
762        assert_eq!(updated.frontier.best_header, frontier(13, 13));
763    }
764
765    #[test]
766    fn stale_header_advance_is_ignored() {
767        let current = update(
768            frontier(5, 5),
769            frontier(8, 8),
770            frontier(12, 12),
771            FrontierChange::Snapshot,
772        );
773        let requested = update(
774            frontier(20, 20),
775            frontier(21, 21),
776            frontier(12, 22),
777            FrontierChange::HeaderAdvanced,
778        );
779
780        assert!(apply_frontier_update(current, requested).is_none());
781    }
782
783    #[test]
784    fn exchange_keeps_latest_value_without_external_receivers() {
785        let initial = update(
786            frontier(0, 0),
787            frontier(0, 0),
788            frontier(0, 0),
789            FrontierChange::Snapshot,
790        );
791        let exchange = ZakuraSyncExchange::new(initial, ZakuraTrace::noop());
792        let accepted = update(
793            frontier(0, 0),
794            frontier(0, 0),
795            frontier(10, 10),
796            FrontierChange::HeaderAdvanced,
797        );
798
799        exchange.publish_frontier(accepted, "test");
800
801        assert_eq!(exchange.current_frontier(), accepted);
802
803        let stale = update(
804            frontier(0, 0),
805            frontier(0, 0),
806            frontier(9, 9),
807            FrontierChange::HeaderAdvanced,
808        );
809        exchange.publish_frontier(stale, "test");
810
811        assert_eq!(exchange.current_frontier(), accepted);
812    }
813
814    #[test]
815    fn concurrent_publishers_converge_to_highest_frontiers() {
816        const PUBLISHERS: u32 = 64;
817
818        let initial = update(
819            frontier(0, 0),
820            frontier(0, 0),
821            frontier(0, 0),
822            FrontierChange::Snapshot,
823        );
824        let exchange = ZakuraSyncExchange::new(initial, ZakuraTrace::noop());
825        let barrier = Arc::new(Barrier::new((PUBLISHERS * 2 + 1) as usize));
826        let mut publishers = Vec::new();
827
828        for height in 1..=PUBLISHERS {
829            let seed = u8::try_from(height).expect("test publisher height fits in u8");
830            let header_exchange = exchange.clone();
831            let header_barrier = barrier.clone();
832            publishers.push(thread::spawn(move || {
833                header_barrier.wait();
834                header_exchange.publish_frontier(
835                    update(
836                        frontier(0, 0),
837                        frontier(0, 0),
838                        frontier(height, seed),
839                        FrontierChange::HeaderAdvanced,
840                    ),
841                    "test",
842                );
843            }));
844
845            let seed = u8::try_from(height).expect("test publisher height fits in u8");
846            let grow_exchange = exchange.clone();
847            let grow_barrier = barrier.clone();
848            publishers.push(thread::spawn(move || {
849                grow_barrier.wait();
850                grow_exchange.publish_frontier(
851                    update(
852                        frontier(height, seed),
853                        frontier(height, seed),
854                        frontier(0, 0),
855                        FrontierChange::VerifiedGrow,
856                    ),
857                    "test",
858                );
859            }));
860        }
861
862        barrier.wait();
863        for publisher in publishers {
864            publisher.join().expect("publisher thread should not panic");
865        }
866
867        let current = exchange.current_frontier().frontier;
868        let seed = u8::try_from(PUBLISHERS).expect("test publisher height fits in u8");
869        assert_eq!(current.finalized, frontier(PUBLISHERS, seed));
870        assert_eq!(current.verified_body, frontier(PUBLISHERS, seed));
871        assert_eq!(current.best_header, frontier(PUBLISHERS, seed));
872    }
873
874    #[test]
875    fn concurrent_header_advance_is_not_lost_to_verified_grow() {
876        let initial = update(
877            frontier(0, 0),
878            frontier(5, 5),
879            frontier(10, 10),
880            FrontierChange::Snapshot,
881        );
882        let exchange = ZakuraSyncExchange::new(initial, ZakuraTrace::noop());
883        let barrier = Arc::new(Barrier::new(3));
884
885        let header_exchange = exchange.clone();
886        let header_barrier = barrier.clone();
887        let header = thread::spawn(move || {
888            header_barrier.wait();
889            header_exchange.publish_frontier(
890                update(
891                    frontier(0, 0),
892                    frontier(0, 0),
893                    frontier(20, 20),
894                    FrontierChange::HeaderAdvanced,
895                ),
896                "test",
897            );
898        });
899
900        let grow_exchange = exchange.clone();
901        let grow_barrier = barrier.clone();
902        let grow = thread::spawn(move || {
903            grow_barrier.wait();
904            grow_exchange.publish_frontier(
905                update(
906                    frontier(0, 0),
907                    frontier(8, 8),
908                    frontier(0, 0),
909                    FrontierChange::VerifiedGrow,
910                ),
911                "test",
912            );
913        });
914
915        barrier.wait();
916        header.join().expect("header publisher should not panic");
917        grow.join().expect("grow publisher should not panic");
918
919        let current = exchange.current_frontier().frontier;
920        assert_eq!(current.verified_body, frontier(8, 8));
921        assert_eq!(current.best_header, frontier(20, 20));
922    }
923}
zakura_network/zakura/exchange.rs

zakura_network/zakura/
exchange.rs
