durable-actors 0.12.2

Standalone regional durable-actors control plane, host, and durability runtime
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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

use super::*;
use crate::bucket::RapidPlacement;
use futures_util::{StreamExt, stream::FuturesUnordered};

type Opened = Vec<(Replica, Box<dyn LogWriter>)>;

pub(super) struct Prepared {
    pub id: String,
    pub prefix: String,
    opened: AbortOnDropHandle<Result<Option<Opened>>>,
}

impl Prepared {
    pub fn new(prefix: String, zones: Vec<Arc<dyn LogZone>>, placement: RapidPlacement) -> Self {
        let id = uuid::Uuid::new_v4().to_string();
        let object = format!("{prefix}{id}.segment");
        let opened = AbortOnDropHandle::new(tokio::spawn(async move {
            match placement {
                RapidPlacement::All => open_all(&zones, &object).await,
                RapidPlacement::Nearest => open_nearest(zones, object).await,
            }
        }));
        Self { id, prefix, opened }
    }

    pub async fn take(self) -> Result<Option<Opened>> {
        self.opened
            .await
            .context("Rapid connection preparation failed")?
    }
}

async fn open_all(zones: &[Arc<dyn LogZone>], object: &str) -> Result<Option<Opened>> {
    let results =
        futures_util::future::join_all(zones.iter().map(|zone| bounded(zone.open(object)))).await;
    let mut opened = Vec::new();
    for result in results {
        match result {
            Ok(copy) => opened.push(copy),
            Err(error) => {
                tracing::warn!(%error, "Rapid log unavailable; activation will use Standard")
            }
        }
    }
    if opened.len() == zones.len() {
        return Ok(Some(opened));
    }
    for (replica, writer) in opened {
        drop(writer);
        if let Some(zone) = zones.iter().find(|zone| zone.bucket() == replica.bucket) {
            let _ = bounded(zone.delete(&replica)).await;
        }
    }
    Ok(None)
}

/// The first zone to open is the closest one from this host; later opens are discarded.
async fn open_nearest(zones: Vec<Arc<dyn LogZone>>, object: String) -> Result<Option<Opened>> {
    let mut opens: FuturesUnordered<_> = zones
        .into_iter()
        .map(|zone| {
            let object = object.clone();
            async move {
                let result = bounded(zone.open(&object)).await;
                (zone, result)
            }
        })
        .collect();
    while let Some((_, result)) = opens.next().await {
        match result {
            Ok(copy) => {
                tokio::spawn(discard(opens));
                return Ok(Some(vec![copy]));
            }
            Err(error) => tracing::warn!(%error, "Rapid zone unavailable during placement"),
        }
    }
    tracing::warn!("Rapid log unavailable; activation will use Standard");
    Ok(None)
}

async fn discard<F>(mut opens: FuturesUnordered<F>)
where
    F: std::future::Future<Output = (Arc<dyn LogZone>, Result<(Replica, Box<dyn LogWriter>)>)>,
{
    while let Some((zone, result)) = opens.next().await {
        if let Ok((replica, writer)) = result {
            drop(writer);
            if let Err(error) = bounded(zone.delete(&replica)).await {
                tracing::warn!(%error, "unused Rapid placement was not deleted");
            }
        }
    }
}