durable-actors 0.3.0

Standalone regional durable-actors control plane, host, and durability runtime
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

use super::*;

impl RuntimeStorage {
    pub(crate) async fn register_initial_replicas(
        &self,
        scope: &ReplicaScope,
        replicas: Vec<ReplicaTarget>,
    ) -> Result<ReplicaMembership> {
        self.validate_initial_replicas(&replicas)?;
        let key = format!(
            "{}.json",
            crate::storage_paths::session(&scope.host, &scope.session)
        );
        let initial = session::Session {
            id: scope.identity(),
            region: scope.region.clone(),
            replicas,
            state: session::RecoveryState::Open,
        };
        let persisted = if replace(
            self.authority.as_ref(),
            &key,
            None,
            serde_json::to_vec(&initial)?,
        )
        .await?
        {
            initial
        } else {
            let object = self
                .authority
                .get(&key)
                .await?
                .context("replication session disappeared")?;
            serde_json::from_slice::<session::Session>(&object.bytes)?
        };
        ensure!(
            persisted.is_open()
                && persisted.id == scope.identity()
                && persisted.region == scope.region,
            "initial replication session changed or was fenced"
        );
        self.validate_initial_replicas(&persisted.replicas)?;
        Ok(ReplicaMembership {
            scope: scope.clone(),
            replicas: persisted.replicas,
        })
    }

    fn validate_initial_replicas(&self, replicas: &[ReplicaTarget]) -> Result<()> {
        ensure!(
            replicas.len() == self.fleet.replica_regions().len(),
            "incomplete initial replica set"
        );
        ReplicationTicket {
            replicas: replicas.to_vec(),
        }
        .validate()
    }
}