commonware_glue/dkg/
state_sync.rs

1//! Durable startup material for DKG actors after application state sync.
2//!
3//! A state-synced node may not have the finalized boundary block that introduced
4//! its active epoch. [`Plan`] persists that public material before actors start
5//! and shares one recovery decision between the reshare and orchestrator actors.
6//! A later startup removes the record after marshal's recovered epoch advances
7//! beyond the synced epoch.
8
9use crate::dkg::{network::Directory, types::EpochInfo};
10use bytes::{Buf, BufMut};
11use commonware_codec::{Decode as _, Encode as _, EncodeSize, Error as CodecError, Read, Write};
12use commonware_consensus::{
13    Epochable as _,
14    marshal::core::{Mailbox as MarshalMailbox, Variant as MarshalVariant},
15    simplex::{scheme::Scheme, types::Finalization},
16    types::{Epoch, Epocher, FixedEpocher},
17};
18#[cfg(feature = "arbitrary")]
19use commonware_cryptography::bls12381::dkg::feldman_desmedt::Output;
20use commonware_cryptography::{
21    Digest,
22    bls12381::primitives::{sharing::ModeVersion, variant::Variant},
23};
24use commonware_storage::{
25    Context,
26    metadata::{self, Metadata},
27};
28use commonware_utils::{
29    fixed_bytes,
30    sequence::{FixedBytes, Unit},
31    sync::AsyncMutex,
32};
33use std::{fmt, num::NonZeroU32, sync::Arc};
34
35const STATE_SYNC_KEY: FixedBytes<1> = fixed_bytes!("00");
36const STATE_SYNC_SUFFIX: &str = "_dkg_state_sync";
37type EpochInfoCodecConfig = (NonZeroU32, ModeVersion);
38
39/// Storage settings for a DKG state-sync recovery plan.
40#[derive(Clone, Debug)]
41pub struct Config {
42    /// Stable node-wide partition prefix.
43    pub partition_prefix: String,
44
45    /// Maximum entries accepted in each persisted participant set.
46    pub max_participants: NonZeroU32,
47
48    /// Maximum sharing mode version accepted in persisted epoch information.
49    pub max_supported_mode: ModeVersion,
50}
51
52/// Public material needed to start DKG actors in a state-synced epoch.
53///
54/// The epoch info replaces the boundary block that may be absent locally. The
55/// floor is the finalized block selected for application state sync and gives
56/// marshal a block from which to resume delivery.
57///
58/// The probe fixes the floor and the epoch info atomically, so the info
59/// always describes the epoch containing the floor.
60pub struct StateSync<S, D, V, Dir = Unit>
61where
62    S: Scheme<D>,
63    D: Digest,
64    V: Variant,
65    Dir: Directory<S::PublicKey>,
66{
67    /// Public information for the epoch containing the state-sync floor.
68    ///
69    /// Carries the epoch's transport directory, so a state-synced node can
70    /// activate the epoch's peers without any application state.
71    pub info: EpochInfo<V, S::PublicKey, Dir>,
72
73    /// Finalized floor selected for application state sync.
74    pub floor: Finalization<S, D>,
75}
76
77impl<S, D, V, Dir> Clone for StateSync<S, D, V, Dir>
78where
79    S: Scheme<D>,
80    D: Digest,
81    V: Variant,
82    Dir: Directory<S::PublicKey>,
83{
84    fn clone(&self) -> Self {
85        Self {
86            info: self.info.clone(),
87            floor: self.floor.clone(),
88        }
89    }
90}
91
92impl<S, D, V, Dir> fmt::Debug for StateSync<S, D, V, Dir>
93where
94    S: Scheme<D>,
95    D: Digest,
96    V: Variant,
97    Dir: Directory<S::PublicKey>,
98{
99    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
100        formatter
101            .debug_struct("StateSync")
102            .field("info", &self.info)
103            .field("floor", &self.floor)
104            .finish()
105    }
106}
107
108impl<S, D, V, Dir> PartialEq for StateSync<S, D, V, Dir>
109where
110    S: Scheme<D>,
111    D: Digest,
112    V: Variant,
113    Dir: Directory<S::PublicKey>,
114{
115    fn eq(&self, other: &Self) -> bool {
116        self.info == other.info && self.floor == other.floor
117    }
118}
119
120impl<S, D, V, Dir> Eq for StateSync<S, D, V, Dir>
121where
122    S: Scheme<D>,
123    D: Digest,
124    V: Variant,
125    Dir: Directory<S::PublicKey>,
126{
127}
128
129impl<S, D, V, Dir> Write for StateSync<S, D, V, Dir>
130where
131    S: Scheme<D>,
132    D: Digest,
133    V: Variant,
134    Dir: Directory<S::PublicKey>,
135{
136    fn write(&self, writer: &mut impl BufMut) {
137        self.info.write(writer);
138        self.floor.write(writer);
139    }
140}
141
142impl<S, D, V, Dir> EncodeSize for StateSync<S, D, V, Dir>
143where
144    S: Scheme<D>,
145    D: Digest,
146    V: Variant,
147    Dir: Directory<S::PublicKey>,
148{
149    fn encode_size(&self) -> usize {
150        self.info.encode_size() + self.floor.encode_size()
151    }
152}
153
154impl<S, D, V, Dir> Read for StateSync<S, D, V, Dir>
155where
156    S: Scheme<D>,
157    D: Digest,
158    V: Variant,
159    Dir: Directory<S::PublicKey>,
160{
161    type Cfg = (EpochInfoCodecConfig, <S::Certificate as Read>::Cfg);
162
163    fn read_cfg(
164        reader: &mut impl Buf,
165        (epoch_info, certificate): &Self::Cfg,
166    ) -> Result<Self, CodecError> {
167        Ok(Self {
168            info: EpochInfo::read_cfg(reader, epoch_info)?,
169            floor: Finalization::read_cfg(reader, certificate)?,
170        })
171    }
172}
173
174#[cfg(feature = "arbitrary")]
175impl<S, D, V, Dir> arbitrary::Arbitrary<'_> for StateSync<S, D, V, Dir>
176where
177    S: Scheme<D>,
178    D: Digest + for<'a> arbitrary::Arbitrary<'a>,
179    V: Variant,
180    Dir: Directory<S::PublicKey> + for<'a> arbitrary::Arbitrary<'a>,
181    S::PublicKey: for<'a> arbitrary::Arbitrary<'a>,
182    S::Certificate: for<'a> arbitrary::Arbitrary<'a>,
183    Output<V, S::PublicKey>: for<'a> arbitrary::Arbitrary<'a>,
184{
185    fn arbitrary(u: &mut arbitrary::Unstructured<'_>) -> arbitrary::Result<Self> {
186        Ok(Self {
187            info: u.arbitrary()?,
188            floor: u.arbitrary()?,
189        })
190    }
191}
192
193/// Returns the epoch containing marshal's next unprocessed height.
194pub(crate) async fn recovered_epoch<S, V>(
195    marshal: &MarshalMailbox<S, V>,
196    epocher: &FixedEpocher,
197) -> Option<Epoch>
198where
199    S: commonware_cryptography::certificate::Scheme,
200    V: MarshalVariant,
201{
202    let height = marshal.get_processed_height().await?.next();
203    Some(
204        epocher
205            .containing(height)
206            .expect("epocher must know recovered height")
207            .epoch(),
208    )
209}
210
211enum PlanState<S, D, V, Dir>
212where
213    S: Scheme<D>,
214    D: Digest,
215    V: Variant,
216    Dir: Directory<S::PublicKey>,
217{
218    Pending {
219        candidate: Option<StateSync<S, D, V, Dir>>,
220        partition: String,
221        codec_config: EpochInfoCodecConfig,
222    },
223    Resolved(Option<StateSync<S, D, V, Dir>>),
224}
225
226/// Shared startup recovery plan for DKG actors.
227///
228/// Initialize one plan after obtaining optional application state-sync material
229/// and before starting either DKG actor. Clones share the resolution decision,
230/// so both actors observe the same material while using one durable partition.
231/// Use [`Plan::disabled`] for deployments that never use application state sync.
232pub struct Plan<S, D, V, Dir = Unit>
233where
234    S: Scheme<D>,
235    D: Digest,
236    V: Variant,
237    Dir: Directory<S::PublicKey>,
238{
239    state: Arc<AsyncMutex<PlanState<S, D, V, Dir>>>,
240}
241
242impl<S, D, V, Dir> Clone for Plan<S, D, V, Dir>
243where
244    S: Scheme<D>,
245    D: Digest,
246    V: Variant,
247    Dir: Directory<S::PublicKey>,
248{
249    fn clone(&self) -> Self {
250        Self {
251            state: self.state.clone(),
252        }
253    }
254}
255
256impl<S, D, V, Dir> fmt::Debug for Plan<S, D, V, Dir>
257where
258    S: Scheme<D>,
259    D: Digest,
260    V: Variant,
261    Dir: Directory<S::PublicKey>,
262{
263    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
264        formatter.debug_struct("Plan").finish_non_exhaustive()
265    }
266}
267
268impl<S, D, V, Dir> Plan<S, D, V, Dir>
269where
270    S: Scheme<D>,
271    D: Digest,
272    V: Variant,
273    Dir: Directory<S::PublicKey>,
274{
275    /// Initializes and durably records a DKG state-sync recovery candidate.
276    ///
277    /// Provided material replaces any existing record. Without provided
278    /// material, the existing record is loaded. The store is closed before this
279    /// method returns, making the plan safe to pass to actors immediately.
280    ///
281    /// # Panics
282    ///
283    /// Panics if storage cannot be loaded or synchronized, if provided or
284    /// persisted material has mismatched artifact and floor epochs, or if
285    /// provided material violates the configured codec limits.
286    pub async fn init<E: Context>(
287        context: E,
288        config: Config,
289        provided: Option<StateSync<S, D, V, Dir>>,
290    ) -> Self {
291        let codec_config = (config.max_participants, config.max_supported_mode);
292        if let Some(provided) = &provided {
293            assert_provided(provided, codec_config);
294        }
295        let partition = format!("{}{STATE_SYNC_SUFFIX}", config.partition_prefix);
296        let mut store =
297            open_store::<E, S, D, V, Dir>(context, partition.clone(), codec_config).await;
298        if let Some(provided) = provided {
299            store.put(STATE_SYNC_KEY, provided);
300            store = store
301                .sync()
302                .await
303                .expect("failed to persist DKG state sync metadata");
304        }
305        let candidate = store.get(&STATE_SYNC_KEY).cloned();
306        if let Some(candidate) = &candidate {
307            assert_epoch(candidate);
308        }
309        drop(store);
310
311        Self {
312            state: Arc::new(AsyncMutex::new(PlanState::Pending {
313                candidate,
314                partition,
315                codec_config,
316            })),
317        }
318    }
319
320    /// Creates a plan that never accesses storage and always resolves to none.
321    pub fn disabled() -> Self {
322        Self {
323            state: Arc::new(AsyncMutex::new(PlanState::Resolved(None))),
324        }
325    }
326
327    pub(crate) async fn resolve<E: Context>(
328        &self,
329        context: E,
330        recovered_epoch: Option<Epoch>,
331    ) -> Option<StateSync<S, D, V, Dir>> {
332        let mut state = self.state.lock().await;
333        let (candidate, partition, codec_config) = match &*state {
334            PlanState::Resolved(resolved) => return resolved.clone(),
335            PlanState::Pending {
336                candidate,
337                partition,
338                codec_config,
339            } => (candidate.clone(), partition.clone(), *codec_config),
340        };
341
342        let Some(candidate) = candidate else {
343            *state = PlanState::Resolved(None);
344            return None;
345        };
346        if recovered_epoch.is_none_or(|epoch| epoch <= candidate.floor.epoch()) {
347            *state = PlanState::Resolved(Some(candidate.clone()));
348            return Some(candidate);
349        }
350
351        let mut store = open_store::<E, S, D, V, Dir>(context, partition, codec_config).await;
352        store.remove(&STATE_SYNC_KEY);
353        store
354            .sync()
355            .await
356            .expect("failed to delete stale DKG state sync metadata");
357        *state = PlanState::Resolved(None);
358        None
359    }
360}
361
362async fn open_store<E, S, D, V, Dir>(
363    context: E,
364    partition: String,
365    epoch_info_codec_config: EpochInfoCodecConfig,
366) -> Metadata<E, FixedBytes<1>, StateSync<S, D, V, Dir>>
367where
368    E: Context,
369    S: Scheme<D>,
370    D: Digest,
371    V: Variant,
372    Dir: Directory<S::PublicKey>,
373{
374    Metadata::init(
375        context,
376        metadata::Config {
377            partition,
378            codec_config: (
379                epoch_info_codec_config,
380                S::certificate_codec_config_unbounded(),
381            ),
382        },
383    )
384    .await
385    .expect("failed to load DKG state sync metadata")
386}
387
388fn assert_provided<S, D, V, Dir>(
389    state_sync: &StateSync<S, D, V, Dir>,
390    codec_config: EpochInfoCodecConfig,
391) where
392    S: Scheme<D>,
393    D: Digest,
394    V: Variant,
395    Dir: Directory<S::PublicKey>,
396{
397    assert_epoch(state_sync);
398    StateSync::<S, D, V, Dir>::decode_cfg(
399        state_sync.encode(),
400        &(codec_config, S::certificate_codec_config_unbounded()),
401    )
402    .expect("provided state sync material must satisfy codec config");
403}
404
405fn assert_epoch<S, D, V, Dir>(state_sync: &StateSync<S, D, V, Dir>)
406where
407    S: Scheme<D>,
408    D: Digest,
409    V: Variant,
410    Dir: Directory<S::PublicKey>,
411{
412    assert!(
413        state_sync.info.epoch == state_sync.floor.epoch(),
414        "state sync artifact and floor must be in the same epoch"
415    );
416}
417
418#[cfg(test)]
419mod tests {
420    use super::*;
421    use crate::dkg::{network::Addresses, tests::mocks, types::EpochOutcome};
422    use commonware_consensus::{
423        simplex::types::{Finalize, Proposal},
424        types::{Round, View},
425    };
426    use commonware_cryptography::{
427        Hasher as _, Sha256, Signer as _,
428        bls12381::{dkg::feldman_desmedt::deal, primitives::sharing::Mode},
429        ed25519,
430    };
431    use commonware_p2p::Address;
432    use commonware_parallel::Sequential;
433    use commonware_runtime::{Runner as _, Supervisor as _, deterministic};
434    use commonware_utils::{N3f1, NZU32, TestRng, non_empty, ordered::Set};
435    use std::net::SocketAddr;
436
437    type TestStateSync = StateSync<mocks::TestScheme, mocks::TestDigest, mocks::TestBlsVariant>;
438
439    fn state_sync(context: &mut deterministic::Context, epoch: Epoch) -> TestStateSync {
440        let fixture = mocks::scheme_fixture_n(context, 4);
441        let participants = Set::from_iter_dedup(fixture.participants);
442        let (output, _) = deal::<mocks::TestBlsVariant, _, N3f1>(
443            TestRng::new(epoch.get() + 1),
444            Mode::NonZeroCounter,
445            participants.clone(),
446        )
447        .expect("test DKG output");
448        let proposal = Proposal::new(
449            Round::new(epoch, View::new(1)),
450            View::zero(),
451            Sha256::hash(&[b"state sync floor"]),
452        );
453        let finalizes = fixture
454            .schemes
455            .iter()
456            .map(|scheme| Finalize::sign(scheme, proposal.clone()).expect("test finalize"))
457            .collect::<Vec<_>>();
458        let floor = Finalization::from_finalizes(
459            &fixture.schemes[0],
460            non_empty![@finalizes.iter()],
461            &Sequential,
462        )
463        .expect("test finalization quorum");
464        TestStateSync {
465            info: EpochInfo {
466                outcome: EpochOutcome::Success,
467                epoch,
468                output,
469                players: participants.clone(),
470                next_players: participants,
471                directory: Unit,
472            },
473            floor,
474        }
475    }
476
477    type TestPlan = Plan<mocks::TestScheme, mocks::TestDigest, mocks::TestBlsVariant>;
478
479    fn config(partition: &str) -> Config {
480        Config {
481            partition_prefix: partition.into(),
482            max_participants: NZU32!(16),
483            max_supported_mode: crate::dkg::tests::max_supported_mode(),
484        }
485    }
486
487    async fn plan(
488        context: deterministic::Context,
489        partition: &str,
490        provided: Option<TestStateSync>,
491    ) -> TestPlan {
492        Plan::init(context, config(partition), provided).await
493    }
494
495    fn assert_state_sync(actual: Option<TestStateSync>, expected: &TestStateSync) {
496        assert_eq!(actual.as_ref(), Some(expected));
497    }
498
499    #[test]
500    fn init_persists_before_resolve() {
501        deterministic::Runner::default().start(|mut context| async move {
502            let expected = state_sync(&mut context, Epoch::new(2));
503            let initialized = plan(
504                context.child("initialized"),
505                "persist-before-resolve",
506                Some(expected.clone()),
507            )
508            .await;
509            drop(initialized);
510
511            let reopened = plan(context.child("reopened"), "persist-before-resolve", None).await;
512            assert_state_sync(
513                reopened.resolve(context.child("resolve"), None).await,
514                &expected,
515            );
516        });
517    }
518
519    #[test]
520    fn two_clones_resolve_identically() {
521        deterministic::Runner::default().start(|mut context| async move {
522            let expected = state_sync(&mut context, Epoch::new(2));
523            let first = plan(context.child("init"), "clones", Some(expected.clone())).await;
524            let second = first.clone();
525
526            assert_state_sync(first.resolve(context.child("first"), None).await, &expected);
527            assert_state_sync(
528                second.resolve(context.child("second"), None).await,
529                &expected,
530            );
531        });
532    }
533
534    #[test]
535    fn stale_resolution_deletes_once_and_reopen_sees_none() {
536        deterministic::Runner::default().start(|mut context| async move {
537            let expected = state_sync(&mut context, Epoch::new(2));
538            let first = plan(context.child("init"), "stale", Some(expected)).await;
539            let second = first.clone();
540
541            assert!(
542                first
543                    .resolve(context.child("first"), Some(Epoch::new(3)))
544                    .await
545                    .is_none()
546            );
547            assert!(
548                second
549                    .resolve(context.child("second"), Some(Epoch::new(3)))
550                    .await
551                    .is_none()
552            );
553            drop(first);
554            drop(second);
555
556            let reopened = plan(context.child("reopened"), "stale", None).await;
557            assert!(
558                reopened
559                    .resolve(context.child("reopened_resolve"), None)
560                    .await
561                    .is_none()
562            );
563        });
564    }
565
566    #[test]
567    fn provided_material_overwrites_persisted_record() {
568        deterministic::Runner::default().start(|mut context| async move {
569            let first_value = state_sync(&mut context, Epoch::new(1));
570            let replacement = state_sync(&mut context, Epoch::new(2));
571            drop(plan(context.child("first"), "overwrite", Some(first_value)).await);
572            drop(
573                plan(
574                    context.child("replacement"),
575                    "overwrite",
576                    Some(replacement.clone()),
577                )
578                .await,
579            );
580
581            let reopened = plan(context.child("reopened"), "overwrite", None).await;
582            assert_state_sync(
583                reopened.resolve(context.child("resolve"), None).await,
584                &replacement,
585            );
586        });
587    }
588
589    #[test]
590    #[should_panic(expected = "state sync artifact and floor must be in the same epoch")]
591    fn artifact_beyond_floor_panics_at_init() {
592        deterministic::Runner::default().start(|mut context| async move {
593            let mut mismatched = state_sync(&mut context, Epoch::new(2));
594            mismatched.info.epoch = Epoch::new(3);
595            let _ = plan(context.child("init"), "mismatch", Some(mismatched)).await;
596        });
597    }
598
599    #[test]
600    #[should_panic(expected = "state sync artifact and floor must be in the same epoch")]
601    fn artifact_below_floor_panics_at_init() {
602        deterministic::Runner::default().start(|mut context| async move {
603            let mut mismatched = state_sync(&mut context, Epoch::new(3));
604            mismatched.info.epoch = Epoch::new(2);
605            let _ = plan(context.child("init"), "mismatch-below", Some(mismatched)).await;
606        });
607    }
608
609    #[test]
610    #[should_panic(expected = "provided state sync material must satisfy codec config")]
611    fn oversized_provided_material_panics_at_init() {
612        deterministic::Runner::default().start(|mut context| async move {
613            let provided = state_sync(&mut context, Epoch::new(2));
614            let mut config = config("oversized-provided");
615            config.max_participants = NZU32!(3);
616            let _ = TestPlan::init(context.child("init"), config, Some(provided)).await;
617        });
618    }
619
620    #[test]
621    #[should_panic(expected = "provided state sync material must satisfy codec config")]
622    fn invalid_addressed_directory_panics_at_init() {
623        deterministic::Runner::default().start(|mut context| async move {
624            let StateSync { info, floor } = state_sync(&mut context, Epoch::new(2));
625            let peers = info
626                .participants()
627                .tracked_peers()
628                .union()
629                .into_iter()
630                .chain([ed25519::PrivateKey::from_seed(10_000).public_key()]);
631            let directory = peers
632                .enumerate()
633                .map(|(index, peer)| {
634                    let socket = SocketAddr::from(([127, 0, 0, 1], index as u16 + 1));
635                    (peer, Address::Symmetric(socket))
636                })
637                .collect::<Addresses<_>>();
638            let EpochInfo {
639                outcome,
640                epoch,
641                output,
642                players,
643                next_players,
644                ..
645            } = info;
646            let invalid = StateSync {
647                info: EpochInfo {
648                    outcome,
649                    epoch,
650                    output,
651                    players,
652                    next_players,
653                    directory,
654                },
655                floor,
656            };
657
658            let _ = Plan::<
659                mocks::TestScheme,
660                mocks::TestDigest,
661                mocks::TestBlsVariant,
662                Addresses<mocks::TestPublicKey>,
663            >::init(
664                context.child("init"),
665                config("invalid-directory"),
666                Some(invalid),
667            )
668            .await;
669        });
670    }
671
672    #[test]
673    fn disabled_resolves_none_without_storage() {
674        deterministic::Runner::default().start(|context| async move {
675            let plan = TestPlan::disabled();
676            assert!(plan.resolve(context, None).await.is_none());
677        });
678    }
679}
680
681#[cfg(all(test, feature = "arbitrary"))]
682mod conformance {
683    use super::*;
684    use crate::dkg::{network::Addresses, tests::mocks};
685    use commonware_codec::conformance::CodecConformance;
686
687    commonware_conformance::conformance_tests! {
688        CodecConformance<StateSync<mocks::TestScheme, mocks::TestDigest, mocks::TestBlsVariant>> => 8192,
689        CodecConformance<StateSync<mocks::TestScheme, mocks::TestDigest, mocks::TestBlsVariant, Addresses<mocks::TestPublicKey>>> => 8192,
690    }
691}
commonware_glue/dkg/state_sync.rs

commonware_glue/dkg/
state_sync.rs
