durable-actors 0.14.3

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

use std::collections::BTreeMap;

use super::{LifecycleEvent, LifecycleKind, STALE_AFTER_MS};
use crate::placement::{ActorResidency, ResidencyHistory, ResidencyPeriod};

pub(super) const EVENT_LIMIT: usize = 10_000;

pub(super) fn history(mut events: Vec<LifecycleEvent>, from: u64, to: u64) -> ResidencyHistory {
    let truncated = events.len() > EVENT_LIMIT;
    events.truncate(EVENT_LIMIT);
    let mut instances = BTreeMap::<String, Vec<LifecycleEvent>>::new();
    for event in events {
        instances
            .entry(event.actor.actor_id.clone())
            .or_default()
            .push(event);
    }
    let mut periods = Vec::new();
    for (actor_id, mut events) in instances {
        events.sort_by_key(|event| (event.report.owner_epoch, event.report.sequence));
        events.dedup_by_key(|event| (event.report.owner_epoch, event.report.sequence));
        let mut latest_time = 0;
        events.retain(|event| {
            let ordered = event.report.occurred_at_ms >= latest_time;
            latest_time = latest_time.max(event.report.occurred_at_ms);
            ordered
        });
        for (index, event) in events.iter().enumerate() {
            let start = event.report.occurred_at_ms.max(from);
            let end = events
                .get(index + 1)
                .map_or(to, |next| next.report.occurred_at_ms.min(to));
            let known_until = if event.report.kind == LifecycleKind::Stopped {
                end
            } else {
                event
                    .report
                    .occurred_at_ms
                    .saturating_add(STALE_AFTER_MS)
                    .min(end)
            };
            let status = match (event.report.kind, event.report.resident) {
                (LifecycleKind::Stopped, _) | (_, Some(false)) => ActorResidency::Dormant,
                (_, Some(true)) => ActorResidency::Live,
                _ => ActorResidency::Unknown,
            };
            push(&mut periods, &actor_id, status, start, known_until);
            push(
                &mut periods,
                &actor_id,
                ActorResidency::Unknown,
                start.max(known_until),
                end,
            );
        }
    }
    ResidencyHistory { periods, truncated }
}

fn push(
    periods: &mut Vec<ResidencyPeriod>,
    actor_id: &str,
    status: ActorResidency,
    from_ms: u64,
    to_ms: u64,
) {
    if from_ms >= to_ms {
        return;
    }
    if let Some(previous) = periods.last_mut()
        && previous.actor_id == actor_id
        && previous.status == status
        && previous.to_ms == from_ms
    {
        previous.to_ms = to_ms;
    } else {
        periods.push(ResidencyPeriod {
            actor_id: actor_id.to_owned(),
            status,
            from_ms,
            to_ms,
        });
    }
}