nixfleet_reconciler/
manifest.rs

1//! Pure projection: fleet.resolved + channel context -> RolloutManifest.
2//! Producer (nixfleet-release) and CP (re-derivation) share this fn.
3
4use anyhow::{Result, anyhow};
5use chrono::{DateTime, Utc};
6use nixfleet_proto::{FleetResolved, HostWave, Meta, RolloutBudget, RolloutId, RolloutManifest};
7
8/// RolloutIds the current fleet snapshot expects across all channels. Filters
9/// `host_dispatch_state` snapshots to "this rev's rollouts only" so stale
10/// Converged rollouts from a previous rev don't poison gate evaluation.
11/// Consumed by polling, the deferrals route, and the per-checkin dispatch
12/// pipeline; centralising it keeps the filter consistent. Errored channels
13/// are silently dropped (callers log at their site, this fn stays pure).
14pub fn current_rollout_ids(
15    fleet: &FleetResolved,
16    fleet_resolved_hash: &str,
17) -> std::collections::HashSet<String> {
18    fleet
19        .channels
20        .keys()
21        .filter_map(|ch| {
22            compute_rollout_id_for_channel(fleet, fleet_resolved_hash, ch)
23                .ok()
24                .flatten()
25        })
26        .collect()
27}
28
29/// CP-side rolloutId for a host on `channel`. `Ok(None)` when the channel
30/// has no host with a declared closure. The id is the canonical RFC-0008 §6.3
31/// composite `"{channel}@{channel_ref}"`, deterministic from the projection
32/// inputs; producer and CP derive the same string for the same inputs.
33pub fn compute_rollout_id_for_channel(
34    fleet: &FleetResolved,
35    fleet_resolved_hash: &str,
36    channel: &str,
37) -> Result<Option<String>> {
38    let signed_at = fleet
39        .meta
40        .signed_at
41        .ok_or_else(|| anyhow!("fleet.meta.signedAt is None - cannot project manifest"))?;
42    let ci_commit = fleet.meta.ci_commit.as_deref();
43    let manifest = match project_manifest(
44        fleet,
45        channel,
46        fleet_resolved_hash,
47        signed_at,
48        ci_commit,
49        fleet.meta.signature_algorithm_or_default(),
50    )? {
51        Some(m) => m,
52        None => return Ok(None),
53    };
54    Ok(Some(
55        RolloutId::new(&manifest.channel, &manifest.channel_ref)
56            .as_str()
57            .to_string(),
58    ))
59}
60
61/// Project one channel out of fleet.resolved. `Ok(None)` when no host on
62/// the channel has a `closureHash`. `host_set` sorted for canonical-byte
63/// stability.
64pub fn project_manifest(
65    fleet: &FleetResolved,
66    channel: &str,
67    fleet_resolved_hash: &str,
68    signed_at: DateTime<Utc>,
69    ci_commit: Option<&str>,
70    signature_algorithm: &str,
71) -> Result<Option<RolloutManifest>> {
72    let channel_def = fleet
73        .channels
74        .get(channel)
75        .ok_or_else(|| anyhow!("channel {channel} missing from fleet.channels"))?;
76
77    let policy = fleet
78        .rollout_policies
79        .get(&channel_def.rollout_policy)
80        .ok_or_else(|| {
81            anyhow!(
82                "rollout policy {} for channel {channel} not found in fleet.rolloutPolicies",
83                channel_def.rollout_policy
84            )
85        })?;
86
87    let waves = fleet.waves.get(channel);
88
89    let mut host_set: Vec<HostWave> = Vec::new();
90    for (hostname, host) in fleet.hosts.iter() {
91        if host.channel != channel {
92            continue;
93        }
94        let target_closure = match host.closure_hash.as_ref() {
95            Some(c) => c.clone(),
96            None => continue,
97        };
98        let wave_index: u32 = match waves {
99            Some(ws) => ws
100                .iter()
101                .position(|w| w.hosts.iter().any(|h| h == hostname))
102                .map(|i| i as u32)
103                .unwrap_or(0),
104            None => 0,
105        };
106        host_set.push(HostWave {
107            hostname: hostname.clone(),
108            wave_index,
109            target_closure,
110        });
111    }
112
113    if host_set.is_empty() {
114        return Ok(None);
115    }
116    host_set.sort_by(|a, b| a.hostname.cmp(&b.hostname));
117
118    let display_name = format!(
119        "{}@{}",
120        channel,
121        ci_commit
122            .map(|c| c.chars().take(8).collect::<String>())
123            .unwrap_or_else(|| "unknown".to_string())
124    );
125
126    let channel_ref = ci_commit.unwrap_or_default().to_string();
127
128    // Snapshot disruption budgets here; selectors resolve once and freeze.
129    // Mid-rollout retags affect future rollouts, never this one. Hosts sorted
130    // for JCS canonical-byte stability.
131    let disruption_budgets: Vec<RolloutBudget> = fleet
132        .disruption_budgets
133        .iter()
134        .map(|b| {
135            let mut hosts = b.selector.resolve(fleet.hosts.iter());
136            hosts.sort();
137            RolloutBudget {
138                selector: b.selector.clone(),
139                hosts,
140                max_in_flight: b.max_in_flight,
141                max_in_flight_pct: b.max_in_flight_pct,
142            }
143        })
144        .collect();
145
146    Ok(Some(RolloutManifest {
147        schema_version: 1,
148        display_name,
149        channel: channel.to_string(),
150        channel_ref,
151        fleet_resolved_hash: fleet_resolved_hash.to_string(),
152        host_set,
153        health_gate: policy.health_gate.clone(),
154        disruption_budgets,
155        meta: Meta {
156            schema_version: 1,
157            signed_at: Some(signed_at),
158            ci_commit: ci_commit.map(|c| c.to_string()),
159            signature_algorithm: Some(signature_algorithm.to_string()),
160        },
161    }))
162}
nixfleet_reconciler/manifest.rs

nixfleet_reconciler/
manifest.rs
