bynk/
migrate.rs

1//! `bynk migrate`: where each agent class's stored state stands, and the sweep
2//! that migrates every stored instance and raises the class's floor
3//! (state-migrations S4, #2259; the retired track's summary in
4//! `design/archive/retired-tracks.md`, §3.3, §3.6).
5//!
6//! `status` is offline. It reads `bynk.schema.lock`, checks the source against
7//! it (read-only, as `bynk check` does), and reads `bynk.deploy.lock` only for
8//! the environments each Worker is pushed in.
9
10use std::collections::BTreeMap;
11use std::path::Path;
12use std::process::ExitCode;
13
14use bynk_emit::project::AgentVersions;
15use bynk_project::schema_registry::SchemaRegistry;
16use serde::Serialize;
17
18pub(crate) mod live;
19mod sweep;
20
21/// Output choices for `bynk migrate status`.
22#[derive(Copy, Clone, Debug, PartialEq, Eq)]
23pub enum StatusFormat {
24    Short,
25    Json,
26}
27
28/// One agent class's migration state.
29#[derive(Debug, PartialEq, Eq, Serialize)]
30pub struct ClassStatus {
31    /// `<context>.<Agent>`, the lock's key.
32    pub class: String,
33    pub version: i64,
34    pub floor: i64,
35    /// The oldest stored version that reads forward to `version` without a
36    /// step (an all-additive path).
37    pub oldest_readable: i64,
38    pub steps: Vec<StepStatus>,
39    /// Versions from the floor up that no step covers and that have no
40    /// all-additive path: a stored record there faults `NoMigrationPath`.
41    pub without_path: Vec<i64>,
42    /// The environments `bynk.deploy.lock` records this class's Worker in. A
43    /// floor-raising sweep must cover every one.
44    pub deployed_to: Vec<String>,
45}
46
47/// One valid `migrate from N` step and the stored versions it covers.
48#[derive(Debug, PartialEq, Eq, Serialize)]
49pub struct StepStatus {
50    pub from: i64,
51    /// The oldest version the step covers (`lo..=from`).
52    pub covers_from: i64,
53}
54
55/// The Worker directory holding agent class `class` (`<unit>.<Agent>`): the
56/// unit's qualified name with `.` replaced by `-`.
57pub(crate) fn worker_of(class: &str) -> String {
58    let unit = class.rsplit_once('.').map_or(class, |(unit, _)| unit);
59    unit.replace('.', "-")
60}
61
62/// Every recorded agent class's status, in key order. `versions` is the
63/// check's reconciled view (steps that hold, by class); `deployed` maps each
64/// Worker directory to its recorded environments.
65pub fn class_statuses(
66    lock: &SchemaRegistry,
67    versions: &std::collections::HashMap<String, AgentVersions>,
68    deployed: &BTreeMap<String, Vec<String>>,
69) -> Vec<ClassStatus> {
70    lock.agents()
71        .map(|(class, entry)| {
72            // A class the source no longer declares has no reconciled view:
73            // its recorded entry alone says where it stands.
74            let (oldest_readable, steps) = match versions.get(class) {
75                Some(v) => (
76                    v.oldest_readable,
77                    v.steps
78                        .iter()
79                        .map(|s| StepStatus {
80                            from: s.from,
81                            covers_from: s.lo,
82                        })
83                        .collect(),
84                ),
85                None => (entry.oldest_readable(), Vec::new()),
86            };
87            let without_path = (entry.floor..entry.version)
88                .filter(|v| {
89                    *v < oldest_readable
90                        && !steps
91                            .iter()
92                            .any(|s: &StepStatus| s.covers_from <= *v && *v <= s.from)
93                })
94                .collect();
95            ClassStatus {
96                class: class.clone(),
97                version: entry.version,
98                floor: entry.floor,
99                oldest_readable,
100                steps,
101                without_path,
102                deployed_to: deployed.get(&worker_of(class)).cloned().unwrap_or_default(),
103            }
104        })
105        .collect()
106}
107
108/// `status`'s short form: one block per class.
109pub fn render_short(classes: &[ClassStatus]) -> String {
110    if classes.is_empty() {
111        return "no agent classes recorded in bynk.schema.lock\n".to_string();
112    }
113    let mut out = String::new();
114    for c in classes {
115        out.push_str(&format!(
116            "{}  version {}, floor {}\n",
117            c.class, c.version, c.floor
118        ));
119        for s in &c.steps {
120            let range = if s.covers_from == s.from {
121                s.from.to_string()
122            } else {
123                format!("{}..={}", s.covers_from, s.from)
124            };
125            out.push_str(&format!(
126                "  migrate from {}  covers {range}; deletable once the floor passes {}\n",
127                s.from, s.from
128            ));
129        }
130        if c.oldest_readable < c.version {
131            out.push_str(&format!(
132                "  versions {}..={} read forward without a step\n",
133                c.oldest_readable,
134                c.version - 1
135            ));
136        }
137        for v in &c.without_path {
138            out.push_str(&format!(
139                "  version {v} has no path: a stored record there faults `NoMigrationPath`\n"
140            ));
141        }
142        if c.deployed_to.is_empty() {
143            out.push_str("  deployed to: nothing recorded in bynk.deploy.lock\n");
144        } else {
145            out.push_str(&format!(
146                "  deployed to: {} (a sweep of {} can raise the floor)\n",
147                c.deployed_to.join(", "),
148                if c.deployed_to.len() == 1 {
149                    "it"
150                } else if c.deployed_to.len() == 2 {
151                    "both"
152                } else {
153                    "all of them"
154                }
155            ));
156        }
157    }
158    out
159}
160
161/// Run `bynk migrate status` on the project at `root`.
162pub fn status(root: &Path, format: StatusFormat) -> ExitCode {
163    match status_report(root) {
164        Ok(classes) => {
165            match format {
166                StatusFormat::Short => print!("{}", render_short(&classes)),
167                StatusFormat::Json => println!(
168                    "{}",
169                    serde_json::to_string_pretty(&serde_json::json!({ "classes": classes }))
170                        .expect("a status report always serializes")
171                ),
172            }
173            ExitCode::SUCCESS
174        }
175        Err(e) => {
176            eprintln!("bynk migrate status: {e}");
177            ExitCode::FAILURE
178        }
179    }
180}
181
182/// The lock, the source checked against it, and the ledger, as statuses. A
183/// source with errors is refused: a step that fails its check covers nothing,
184/// so the report would understate what the next build does.
185fn status_report(root: &Path) -> Result<Vec<ClassStatus>, String> {
186    let existing = bynk_driver::schema_lock::read(root).map_err(|e| {
187        format!(
188            "could not read {}: {e}",
189            bynk_driver::schema_lock::lock_path(root).display()
190        )
191    })?;
192    let lock = bynk_project::schema_registry::parse(existing.as_deref(), root)?;
193    let options = bynk_driver::try_project_options(root).map_err(|e| e.to_string())?;
194    let options = options.schema_registry(bynk_emit::project::SchemaLock::On { existing });
195    let check = bynk_emit::project::check_project(&options);
196    if check.has_errors() {
197        bynk_driver::print_project_check_short(&check);
198        return Err("the project has errors; fix them first (`bynk check`)".to_string());
199    }
200    let workers: Vec<String> = lock.agents().map(|(class, _)| worker_of(class)).collect();
201    let deployed = crate::deploy::deployed_environments(root, &workers)?;
202    Ok(class_statuses(&lock, &check.agent_versions, &deployed))
203}
204
205/// `bynk migrate sweep`'s options.
206#[derive(Debug, Clone, Default)]
207pub struct SweepOptions {
208    /// Only this context's classes (its qualified name or Worker directory).
209    pub context: Option<String>,
210    /// Only these environments; every recorded one when empty. A narrowed
211    /// run migrates but never raises a floor.
212    pub environments: Vec<String>,
213    /// Instances woken per Worker request.
214    pub batch: usize,
215    /// Skip the confirmation.
216    pub yes: bool,
217    /// `<deployed Worker name>=<https://its host>`, for a Worker that does
218    /// not serve on `workers.dev`.
219    pub urls: Vec<String>,
220}
221
222/// Run `bynk migrate sweep` on the project at `root`.
223pub fn sweep(root: &Path, opts: &SweepOptions) -> ExitCode {
224    match sweep_run(root, opts) {
225        Ok(true) => ExitCode::SUCCESS,
226        Ok(false) => ExitCode::FAILURE,
227        Err(e) => {
228            eprintln!("bynk migrate sweep: {e}");
229            ExitCode::FAILURE
230        }
231    }
232}
233
234/// The sweep, end to end: `Ok(true)` when every selected instance woke.
235fn sweep_run(root: &Path, opts: &SweepOptions) -> Result<bool, String> {
236    let mut urls = BTreeMap::new();
237    for entry in &opts.urls {
238        let (script, url) = entry
239            .split_once('=')
240            .filter(|(s, u)| !s.is_empty() && u.starts_with("https://"))
241            .ok_or_else(|| format!("`--url {entry}`: expected `<worker>=https://<host>`"))?;
242        urls.insert(script.to_string(), url.to_string());
243    }
244    let existing = bynk_driver::schema_lock::read(root).map_err(|e| e.to_string())?;
245    let lock = bynk_project::schema_registry::parse(existing.as_deref(), root)?;
246    let workers: Vec<String> = lock.agents().map(|(class, _)| worker_of(class)).collect();
247    let deployed = crate::deploy::deployed_environments(root, &workers)?;
248    let selection = sweep::select(
249        &lock,
250        &deployed,
251        opts.context.as_deref(),
252        &opts.environments,
253    )?;
254    for class in &selection.never_deployed {
255        eprintln!("{class}: its Worker is recorded in no environment; nothing to sweep");
256    }
257    if selection.targets.is_empty() {
258        return Err("nothing to sweep".to_string());
259    }
260    let credentials = live::Credentials::from_env()?;
261    let agent = live::agent()?;
262    let rest = live::Rest {
263        agent: agent.clone(),
264        credentials,
265        base: live::API.to_string(),
266    };
267    // #2276: the listing's age is measured from when it is taken.
268    let listed_at = std::time::SystemTime::now();
269    let plans = sweep::prepare(&selection, &rest, &urls)?;
270    let instances: usize = plans
271        .values()
272        .flat_map(|p| p.classes.iter().map(|(_, ids)| ids.len()))
273        .sum();
274    for (script, plan) in &plans {
275        for (target, ids) in &plan.classes {
276            eprintln!(
277                "{script} ({}): {}  {} instances with stored data",
278                target.environment,
279                target.agent,
280                ids.len()
281            );
282        }
283    }
284    eprintln!(
285        "The sweep sets a temporary secret, {}, on {} Worker(s) and deletes it afterwards; \
286         each set and delete creates a Worker version. It wakes {instances} instance(s).",
287        bynk_emit::wire_format::MIGRATE_KEY,
288        plans.len()
289    );
290    if !confirm(opts.yes) {
291        return Ok(false);
292    }
293    let probe = crate::probe::detect(
294        &crate::probe::SystemToolbox,
295        "wrangler",
296        crate::probe::DetectOpts {
297            project_root: Some(root),
298            allow_npx: true,
299        },
300    );
301    let secrets = live::WranglerSecrets {
302        provenance: probe.provenance,
303        scratch: std::env::temp_dir().join(format!("bynk-migrate-{}", std::process::id())),
304    };
305    let results = {
306        // A Ctrl-C stops the waking, never the key's deletion.
307        let _deferred = crate::stop::defer();
308        sweep::execute(
309            &plans,
310            &secrets,
311            &live::Https(agent),
312            opts.batch,
313            &sweep::Timing::LIVE,
314            &crate::deploy::mint_key,
315            &|| crate::stop::received().is_some(),
316            &mut std::io::stdout(),
317        )
318    };
319    let _ = std::fs::remove_dir_all(&secrets.scratch);
320    let unsettled = sweep::unsettled(&plans, listed_at, sweep::SETTLING_MARGIN);
321    let raises = sweep::floor_raises(&lock, &selection, &results, &unsettled);
322    // The classes whose floor would have risen had every Worker settled: only
323    // their held floor is the margin's doing (#2279 review).
324    let would_rise: Vec<String> =
325        sweep::floor_raises(&lock, &selection, &results, &BTreeMap::new())
326            .into_iter()
327            .map(|(class, _, _)| class)
328            .collect();
329    if !raises.is_empty() {
330        let mut lock = lock;
331        for (class, from, to) in &raises {
332            lock.raise_floor(class, *to)?;
333            println!("floor: {class} {from} → {to} (every recorded environment swept in this run)");
334        }
335        bynk_driver::schema_lock::write(root, &bynk_project::schema_registry::serialize(&lock))
336            .map_err(|e| format!("could not write bynk.schema.lock: {e}"))?;
337        println!(
338            "bynk.schema.lock updated; commit it. The next build warns about every `migrate \
339             from N` step below the new floor (`bynk.agent.migrate_below_floor`): delete them."
340        );
341    }
342    let raised: Vec<&String> = raises.iter().map(|(c, _, _)| c).collect();
343    for class in selection.covers_all.keys() {
344        if raised.contains(&class) {
345            continue;
346        }
347        if let Some(why) = sweep::floor_held_reason(
348            class,
349            &selection,
350            &results,
351            &unsettled,
352            would_rise.contains(class),
353        ) {
354            println!("floor: {class} unchanged ({why})");
355        }
356    }
357    Ok(results.iter().all(sweep::ClassResult::complete))
358}
359
360/// Ask before changing anything on the account, unless `--yes`.
361fn confirm(yes: bool) -> bool {
362    use std::io::{IsTerminal, Write};
363    if yes {
364        return true;
365    }
366    if !std::io::stdin().is_terminal() {
367        eprintln!("bynk: refusing to change Cloudflare in a non-interactive session without --yes");
368        return false;
369    }
370    eprint!("Sweep? [y/N] ");
371    let _ = std::io::stderr().flush();
372    let mut answer = String::new();
373    std::io::stdin().read_line(&mut answer).is_ok()
374        && matches!(answer.trim().to_ascii_lowercase().as_str(), "y" | "yes")
375}
376
377#[cfg(test)]
378mod tests {
379    use super::*;
380    use bynk_emit::project::MigrationStep;
381    use bynk_project::schema_registry::{AgentEntry, AgentShape};
382
383    fn entry(version: i64, floor: i64, additive: &[bool]) -> AgentEntry {
384        AgentEntry {
385            version,
386            floor,
387            shapes: (floor..=version)
388                .zip(additive.iter())
389                .map(|(v, a)| AgentShape {
390                    version: v,
391                    additive: *a,
392                    fields: Vec::new(),
393                })
394                .collect(),
395            renamed_from: Vec::new(),
396        }
397    }
398
399    fn step(lo: i64, from: i64) -> MigrationStep {
400        MigrationStep {
401            lo,
402            from,
403            carried: Vec::new(),
404            known: Vec::new(),
405            store_maps: Vec::new(),
406        }
407    }
408
409    #[test]
410    fn worker_of_dasherises_the_unit() {
411        assert_eq!(worker_of("shop.Cart"), "shop");
412        assert_eq!(worker_of("acme.shop.Cart"), "acme-shop");
413    }
414
415    #[test]
416    fn a_class_reports_its_steps_gaps_and_environments() {
417        let mut lock = SchemaRegistry::new();
418        // 1 -> 2 additive, 2 -> 3 not, 3 -> 4 not: a step from 3 only.
419        lock.insert_agent(
420            "shop.Cart".to_string(),
421            entry(4, 1, &[false, true, false, false]),
422        );
423        let versions = std::collections::HashMap::from([(
424            "shop.Cart".to_string(),
425            AgentVersions {
426                version: 4,
427                oldest_readable: 4,
428                steps: vec![step(3, 3)],
429            },
430        )]);
431        let deployed = BTreeMap::from([(
432            "shop".to_string(),
433            vec!["default".to_string(), "staging".to_string()],
434        )]);
435        let classes = class_statuses(&lock, &versions, &deployed);
436        assert_eq!(
437            classes,
438            vec![ClassStatus {
439                class: "shop.Cart".to_string(),
440                version: 4,
441                floor: 1,
442                oldest_readable: 4,
443                steps: vec![StepStatus {
444                    from: 3,
445                    covers_from: 3
446                }],
447                without_path: vec![1, 2],
448                deployed_to: vec!["default".to_string(), "staging".to_string()],
449            }]
450        );
451        let text = render_short(&classes);
452        assert!(text.contains("shop.Cart  version 4, floor 1"), "{text}");
453        assert!(text.contains("migrate from 3  covers 3;"), "{text}");
454        assert!(text.contains("version 1 has no path"), "{text}");
455        assert!(
456            text.contains("deployed to: default, staging (a sweep of both"),
457            "{text}"
458        );
459    }
460
461    #[test]
462    fn a_class_gone_from_source_reads_from_its_entry() {
463        let mut lock = SchemaRegistry::new();
464        lock.insert_agent("shop.Old".to_string(), entry(2, 1, &[false, true]));
465        let classes = class_statuses(&lock, &std::collections::HashMap::new(), &BTreeMap::new());
466        assert_eq!(classes[0].oldest_readable, 1);
467        assert!(classes[0].without_path.is_empty());
468        assert!(render_short(&classes).contains("nothing recorded in bynk.deploy.lock"));
469    }
470}
bynk/migrate.rs

bynk/
migrate.rs
