Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions catalog/instances/start_instance_infra/metadata.json
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
{
"type": "StartInstanceInfra",
"label": "Start instance infra",
"description": "Bring up one instance's copy of a `@per_instance` infra node; `done` fires once it answers, so the instance's triggers reading it can be activated next. The run waits without holding a worker; a copy that fails to come up fails the node with the reason; an already-running copy is left as is.",
"description": "Bring up one instance's copy of a `@per_instance` infra node; `done` fires once it answers, so the instance's triggers reading it can be activated next. The run waits without holding a worker; a copy that fails to come up fails the node with the reason; an already-running copy is left as is. If you want to answer a caller fast (a route replying 202), turn `waitUntilRunning` off and reply after this node: it finishes as soon as weft accepts the start, and a status read right after shows the copy as `provisioning`.",
"tags": [
"instances",
"infra",
Expand All @@ -21,13 +21,20 @@
"type": "String",
"required": true,
"description": "The instance id your program gives it (`user-42`)."
},
{
"name": "waitUntilRunning",
"type": "Boolean",
"default": true,
"label": "Wait until running",
"description": "On: `done` fires once the copy runs, and a copy that fails to come up fails the node. Off: `done` fires once weft accepts the start (a start refused for a passing reason, such as a build in progress, is asked again until it is), and whether the copy then comes up is for a status read to tell."
}
],
"outputs": [
{
"name": "done",
"type": "Boolean",
"description": "True once the copy is running."
"description": "True once the copy is running, or once weft accepted the start when `waitUntilRunning` is off."
}
],
"validate": [
Expand Down
11 changes: 9 additions & 2 deletions catalog/instances/start_instance_infra/mod.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
//! StartInstanceInfra: bring up one instance's copy of a `@per_instance`
//! infra node, and fire `done` once the copy answers.
//! infra node, and fire `done` once the copy answers, or, with
//! `waitUntilRunning` off, once weft accepted the start.

use async_trait::async_trait;

Expand All @@ -23,7 +24,13 @@ impl Node for StartInstanceInfraNode {

async fn run(&self, ctx: ExecutionContext) -> WeftResult<()> {
let node: String = ctx.inputs.get("node")?;
ctx.infra(node).instance(instance(&ctx)?).start().await?;
let wait_until_running: bool = ctx.inputs.get("waitUntilRunning")?;
let copy = ctx.infra(node).instance(instance(&ctx)?);
if wait_until_running {
copy.start().await?;
} else {
copy.request_start().await?;
}
ctx.pulse_downstream(NodeOutput::new().set("done", true)).await
}
}
36 changes: 34 additions & 2 deletions catalog/instances/start_instance_infra/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,9 @@
//! copy of that node and `done` waits for the copy to run, a copy that
//! fails fails the node, a copy somebody stops (or that never came to be)
//! while the node waits fails it instead of being started again, a start
//! refused for a passing reason is asked again, and a blank instance is
//! refused before anything is asked.
//! refused for a passing reason is asked again, a node told not to wait
//! finishes once the start is accepted without looking at the copy, and a
//! blank instance is refused before anything is asked.

use serde_json::json;

Expand All @@ -20,6 +21,8 @@ pub fn tests() -> Vec<NodeTest> {
NodeTest::fake("a_copy_stopped_while_waiting_fails_the_node", stopped_while_waiting),
NodeTest::fake("a_start_that_ends_without_a_copy_fails_the_node", no_copy),
NodeTest::fake("a_start_refused_for_now_is_asked_again", refused_for_now),
NodeTest::fake("without_waiting_finishes_once_the_start_is_accepted", without_waiting),
NodeTest::fake("without_waiting_a_start_refused_for_now_is_asked_again", without_waiting_refused_for_now),
NodeTest::fake("a_blank_instance_is_refused", blank_instance),
]
}
Expand Down Expand Up @@ -101,3 +104,32 @@ async fn refused_for_now(rig: FakeRig) -> WeftResult<()> {
assert_eq!(outcome.outputs["done"], json!(true));
Ok(())
}

async fn without_waiting(rig: FakeRig) -> WeftResult<()> {
let outcome = rig
.run(&StartInstanceInfraNode, json!({ "node": "bridge", "instance": "user-42", "waitUntilRunning": false }))
.await
.ok()?;
let calls = rig.program_calls();
assert_eq!(calls.len(), 1, "the start only, no look at the copy: {calls:?}");
assert!(matches!(&calls[0], (ProgramCall::InfraStart { .. }, StopSelf::Keep)), "{calls:?}");
assert!(rig.awaited_signals().is_empty());
assert_eq!(outcome.outputs["done"], json!(true));
Ok(())
}

async fn without_waiting_refused_for_now(rig: FakeRig) -> WeftResult<()> {
rig.answer_program_call("weft.infra.start", json!({ "answer": "waiting", "reason": "project is building" }));
rig.answer_program_call("weft.infra.start", json!({ "answer": "already_starting" }));
rig.signal(json!({}));
let outcome = rig
.run(&StartInstanceInfraNode, json!({ "node": "bridge", "instance": "user-42", "waitUntilRunning": false }))
.await
.ok()?;
let calls = rig.program_calls();
assert_eq!(calls.len(), 2, "asked again after the wait, then done without a look: {calls:?}");
assert!(calls.iter().all(|(call, _)| matches!(call, ProgramCall::InfraStart { .. })), "{calls:?}");
assert_eq!(rig.awaited_signals().len(), 1);
assert_eq!(outcome.outputs["done"], json!(true));
Ok(())
}
11 changes: 10 additions & 1 deletion crates/weft-cli/src/commands/frontend.rs
Original file line number Diff line number Diff line change
Expand Up @@ -154,11 +154,20 @@ fn hand_over(ctx: &Ctx, project: &str, install_url: &str, f: &Frontend, token: O
if let Some(url) = &f.url {
println!("visitors reach it at {url}");
}
let project = ctx.project()?;
println!(
"next, in {}: {} hands its deploy workflow the service and a token",
repo.name,
super::target::export_command(ctx.project()?, ctx.on())
super::target::export_command(project, ctx.on())
);
if !project.root.join(super::target::FRONT_ENV_FILE).is_file() {
println!(
"if the frontend needs settings of its own (its database address, a sign-in secret), write them to \
{} and add `--front-env {}`",
super::target::FRONT_ENV_FILE,
super::target::FRONT_ENV_FILE
);
}
return Ok(());
}
// A frontend running elsewhere reaches the install at its public
Expand Down
10 changes: 4 additions & 6 deletions crates/weft-cli/src/commands/infra.rs
Original file line number Diff line number Diff line change
Expand Up @@ -624,9 +624,8 @@ async fn infra_status(ctx: &Ctx) -> Result<()> {
eprintln!("warning: {note}");
}
// Printed apart from the copies: a start stuck this way has none yet.
let now = crate::progress::now_unix() as i64;
for entry in &status.unanswered {
eprintln!("warning: {}", entry.describe(now));
eprintln!("warning: {}", entry.describe());
}
Ok(())
}
Expand All @@ -644,15 +643,14 @@ async fn changing_now(client: &crate::client::DispatcherClient, project_id: &str
};
// A call of weft's own that keeps failing comes first: it is why
// nothing below moves.
let now = crate::progress::now_unix() as i64;
let unanswered = status.unanswered.iter().map(|u| u.describe(now));
let unanswered = status.unanswered.iter().map(|u| u.describe());
let copies = status.nodes.iter().filter_map(|n| {
let node = match &n.instance {
Some(instance) => format!("{} (instance {instance})", n.node),
None => n.node.clone(),
};
if let Some(progress) = &n.progress {
return Some(format!("{node}: {} {}", n.status, progress.describe_now()));
return Some(format!("{node}: {} {}", n.status, progress.describe(progress.as_of_unix)));
}
let changing = matches!(
InfraNodeStatus::parse(&n.status),
Expand Down Expand Up @@ -695,7 +693,7 @@ fn print_status(name: &str, id: &str, status: &InfraStatus) {
let url = n.endpoint_url.as_deref().unwrap_or("(no endpoint)");
println!(" {node} [{}] -> {url}", n.status);
if let Some(progress) = &n.progress {
println!(" {}", progress.describe_now());
println!(" {}", progress.describe(progress.as_of_unix));
}
// A public endpoint's outside address: what to hand to whoever
// calls in (the node declared only its own path).
Expand Down
7 changes: 3 additions & 4 deletions crates/weft-cli/src/commands/status.rs
Original file line number Diff line number Diff line change
Expand Up @@ -176,7 +176,7 @@ pub async fn run(ctx: Ctx) -> Result<()> {
st => {
println!(" {node}: {st} ({})", entry.endpoint_url.as_deref().unwrap_or("-"));
if let Some(progress) = &entry.progress {
println!(" {}", progress.describe_now());
println!(" {}", progress.describe(progress.as_of_unix));
}
}
}
Expand Down Expand Up @@ -210,7 +210,7 @@ pub async fn run(ctx: Ctx) -> Result<()> {
for entry in &data.instance_infra {
println!(" {} (instance {}): {}", entry.node, entry.instance, entry.status);
if let Some(progress) = &entry.progress {
println!(" {}", progress.describe_now());
println!(" {}", progress.describe(progress.as_of_unix));
}
}
}
Expand All @@ -237,9 +237,8 @@ pub async fn run(ctx: Ctx) -> Result<()> {
// not move.
if !data.unanswered.is_empty() {
println!(" waiting on weft:");
let now = crate::progress::now_unix() as i64;
for entry in &data.unanswered {
println!(" {}", entry.describe(now));
println!(" {}", entry.describe());
}
}
// What each trigger started lately: the one trace a run kept
Expand Down
21 changes: 16 additions & 5 deletions crates/weft-cli/src/commands/target.rs
Original file line number Diff line number Diff line change
Expand Up @@ -376,9 +376,11 @@ async fn export(
for (k, v) in &settings.variables {
println!(" variable {k}={v}");
}
// Only how many: nothing that comes out of a secret reaches the
// terminal, its name included.
println!(" secrets {} set", settings.secrets.len());
// The names only: they are weft's own fixed words, so a person can check
// what the workflow received; a value never reaches the terminal.
for (k, _) in &settings.secrets {
println!(" secret {k}");
}
println!("run its deploy workflow from the Actions tab");
if let Some(frontend) = deployed_before(&minted, frontend) {
println!("frontend '{frontend}' keeps its old token working until that run deploys the new one and retires it");
Expand Down Expand Up @@ -702,9 +704,15 @@ fn committed_target_matches(committed: &str, name: &str, url: &str) -> Result<()
}
}

/// Where a project keeps its frontend server's own settings for a cloud
/// target, the file the next export's `--front-env` hands the deploy
/// workflow: in `front/`, which the workflow builds.
pub const FRONT_ENV_FILE: &str = "front/.env.prod";

/// The `weft target export ... --github` a person runs next, spelled with
/// a target they really have: the one this command was given, else the
/// project's only cloud target, else the choice between them.
/// project's only cloud target, else the choice between them. With
/// [`FRONT_ENV_FILE`] in the project, it hands that file over too.
pub fn export_command(project: &weft_compiler::project::Project, on: Option<&str>) -> String {
let name = match on {
Some(on) => on.to_string(),
Expand All @@ -723,7 +731,10 @@ pub fn export_command(project: &weft_compiler::project::Project, on: Option<&str
}
}
};
format!("`weft target export {name} --github`")
match project.root.join(FRONT_ENV_FILE).is_file() {
true => format!("`weft target export {name} --github --front-env {FRONT_ENV_FILE}`"),
false => format!("`weft target export {name} --github`"),
}
}

/// A target name is what people type after `--on`, and a TOML key.
Expand Down
52 changes: 34 additions & 18 deletions crates/weft-core/src/context/program_calls.rs
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,8 @@ impl InfraCalls<'_> {
}

/// Bring the copy up, and return once it answers, so what comes
/// next (activating the triggers that read it, say) can use it.
/// next (activating the triggers that read it, say) can use it: a
/// [`Self::request_start`], then a wait for the copy to run.
/// Starting a copy that runs, or is already starting, does not start
/// it twice. A copy that fails to come up fails this call with the
/// reason, and so does one somebody stops (or terminates) while this
Expand All @@ -158,25 +159,12 @@ impl InfraCalls<'_> {
/// next to their infra never waits on the run that asked for it.
pub async fn start(self) -> WeftResult<()> {
let instance = instance_id(&self.instance)?;
self.ask_until_accepted(&instance).await?;
let node = self.node;
let start = || ProgramCall::InfraStart { node: node.clone(), instance: instance.clone() };
// Until a start is queued (one refused for a passing reason, a
// build in progress or a trigger of the copy mid-activation, is
// asked again at each look), what the copy reads is no answer to
// this start.
let mut queued = false;
loop {
if !queued {
let answer: InfraStartAnswer = self.ctx.program_call(start(), StopSelf::Keep).await?;
queued = answer.queued();
if !queued {
self.ctx.await_signal(START_LOOK_AGAIN).await?;
continue;
}
}
// A queued start reads `provisioning` from the moment the call
// above answered (the runtime counts the start under way, not
// only what the copy's row says), so every other answer is
// An accepted start reads `provisioning` from the moment the
// runtime accepted it (the runtime counts the start under way,
// not only what the copy's row says), so every other answer is
// where the copy really stands.
let status = ProgramCall::InfraStatus { node: node.clone(), instance: instance.clone() };
let copy: Option<InfraCopy> = self.ctx.program_call(status, StopSelf::Keep).await?;
Expand Down Expand Up @@ -205,6 +193,34 @@ impl InfraCalls<'_> {
}
}

/// Ask for the copy to come up, and return once weft accepted the
/// start, without waiting for the copy to run. From then on a status
/// read ([`Self::status`], [`Self::copies`]) shows the copy as
/// `provisioning`, so a program answering a caller fast (a route
/// replying 202) replies after this and the caller's next read finds
/// the copy. A start refused for a passing reason (a build in
/// progress, a trigger of the copy mid-activation) is asked again
/// every few seconds until it is accepted; a copy already running or
/// starting is not started twice. Whether the copy then comes up is
/// for a later `status` read to tell: nothing here waits for it.
pub async fn request_start(self) -> WeftResult<()> {
let instance = instance_id(&self.instance)?;
self.ask_until_accepted(&instance).await
}

/// Ask for the start until weft accepts it, parking on a timer
/// between refusals for a passing reason.
async fn ask_until_accepted(&self, instance: &Option<InstanceId>) -> WeftResult<()> {
loop {
let call = ProgramCall::InfraStart { node: self.node.clone(), instance: instance.clone() };
let answer: InfraStartAnswer = self.ctx.program_call(call, StopSelf::Keep).await?;
if answer.accepted() {
return Ok(());
}
self.ctx.await_signal(START_LOOK_AGAIN).await?;
}
}

/// Scale the copy down, keeping its disk. `spec` says what happens to
/// the triggers reading it and the runs using it; `stop_self` whether
/// this run goes too when it uses the copy. Answers whether there was
Expand Down
51 changes: 31 additions & 20 deletions crates/weft-core/src/infra/wire.rs
Original file line number Diff line number Diff line change
Expand Up @@ -337,23 +337,24 @@ pub struct InfraStatusEntry {
/// How far a start of an infra copy got: since when it runs, and what it
/// waits on right now, in the host's words ("its machine's agent does not
/// answer yet: ..."). `waiting` is absent before the host reports
/// anything (the machine is still being made).
/// anything (the machine is still being made). `since_unix` is when the
/// start was asked, and `as_of_unix` the server's clock when it answered:
/// both on one clock, so how long it has run never depends on the
/// reader's own.
// SYNC: ApplyProgress <-> packages/weft-graph/src/protocol.ts ApplyProgress
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ApplyProgress {
#[serde(rename = "sinceUnix")]
pub since_unix: i64,
#[serde(rename = "asOfUnix")]
pub as_of_unix: i64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub waiting: Option<String>,
}

impl ApplyProgress {
/// The progress a row records, when it records one.
pub fn of(since_unix: Option<i64>, waiting: Option<String>) -> Option<Self> {
since_unix.map(|since_unix| Self { since_unix, waiting })
}

/// As a person reads it, `now_unix` being the time it is read:
/// As a person reads it, `now_unix` being the time it is read, on the
/// clock `since_unix` is on (`as_of_unix` reads it as of the answer):
/// "for 3m12s, waiting on: ...".
// SYNC: describe <-> packages/weft-graph/src/status.ts describeProgress
pub fn describe(&self, now_unix: i64) -> String {
Expand All @@ -364,15 +365,6 @@ impl ApplyProgress {
None => format!("for {elapsed}"),
}
}

/// [`Self::describe`], read now.
pub fn describe_now(&self) -> String {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(self.since_unix);
self.describe(now)
}
}

#[cfg(test)]
Expand All @@ -381,10 +373,29 @@ mod apply_progress_tests {

#[test]
fn progress_reads_as_how_long_and_on_what() {
let p = ApplyProgress { since_unix: 100, waiting: Some("db: its machine's agent does not answer yet".into()) };
assert_eq!(p.describe(292), "for 3m12s, waiting on: db: its machine's agent does not answer yet");
assert_eq!(ApplyProgress { since_unix: 100, waiting: None }.describe(130), "for 30s");
assert_eq!(ApplyProgress::of(None, Some("x".into())), None, "no start, no progress");
let p = ApplyProgress { since_unix: 100, as_of_unix: 292, waiting: Some("db: its machine's agent does not answer yet".into()) };
assert_eq!(p.describe(p.as_of_unix), "for 3m12s, waiting on: db: its machine's agent does not answer yet");
let bare = ApplyProgress { since_unix: 100, as_of_unix: 130, waiting: None };
assert_eq!(bare.describe(bare.as_of_unix), "for 30s");
}

/// Read as of the answer, the duration is the server's alone: a
/// reader whose clock runs 20s behind the server's still reads 20s.
#[test]
fn progress_as_of_the_answer_ignores_the_readers_clock() {
let p = ApplyProgress { since_unix: 1_000, as_of_unix: 1_020, waiting: None };
assert_eq!(p.describe(p.as_of_unix), "for 20s");
assert_eq!(p.describe(990), "for 0s", "a time before the start reads as no time at all");
}

#[test]
fn progress_travels_as_camel_case() {
let p = ApplyProgress { since_unix: 1, as_of_unix: 2, waiting: None };
assert_eq!(serde_json::to_value(&p).unwrap(), serde_json::json!({ "sinceUnix": 1, "asOfUnix": 2 }));
assert!(
serde_json::from_value::<ApplyProgress>(serde_json::json!({ "sinceUnix": 1 })).is_err(),
"an answer without its clock is refused"
);
}
}

Expand Down
Loading
Loading