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
36 changes: 30 additions & 6 deletions .claude/skills/releasing/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -43,12 +43,36 @@ versions bumped` (PRs only), `build`, `public proxy routes`,
`cargo test + clippy`, `cargo test --features db-tests`, and
`graph + editor`.

## Run the checks CI runs, not the ones you touched

`cargo clippy --workspace --all-targets --locked -- -D warnings` is
the CI line. Running clippy on the crates you edited is not the same
thing and has already sent a PR red on a lint in a crate the change
only touched indirectly.
## Before every push to a PR, run what the PR runs

Before you push a commit to a branch with a PR (the first push that
opens it, and every fix pushed after), run locally everything the PR's
checks run, not only the tests your change reaches. This is the one
place the [test scope] does not apply: a red PR costs a full CI round
trip, and twice a PR has gone red on a test in a crate the change only
touched indirectly. Background each one, and fix everything before the
push, not after.

The PR runs `.github/workflows/ci.yml` and `.github/workflows/docs.yml`;
read them for the current list. At the time of writing:

- `scripts/check-extension-bump.sh origin/main`
- `cargo nextest run --workspace --locked`
- `cargo test --workspace --locked --doc`
- `cargo clippy --workspace --all-targets --locked -- -D warnings`
- `cargo check -p weft-compiler --no-default-features --locked`
- `scripts/validate-examples.sh`
- `scripts/run-db-tests.sh` (every crate, no filter)
- `pnpm -C packages/weft-graph test`, `pnpm -C packages/weft-connect test`,
`pnpm -C packages/weft-syntax test`, `pnpm -C extension-vscode test`,
`pnpm -C extension-vscode run compile`
- `node --test catalog/bailey/bridge/images/bridge/src/*.test.js`
- `pnpm -C extension-browser exec tsc --noEmit`, `pnpm -C extension-browser check`
- the docs build: `cd docs && mdbook-mermaid install . && mdbook build`

Clippy on the crates you edited is not the same as the workspace line
above, and has sent a PR red on a lint in a crate the change only
touched indirectly.

`scripts/validate-examples.sh` compiles every example. It fails on
ERRORS only, so a warning is advice an example may carry. If an
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@
},
"default": "park",
"label": "Triggers",
"description": "park queues what arrives, hibernate keeps accepting for `graceMinutes`, wipe drops it."
"description": "Park keeps everything that arrives (calls, answers, the trigger's own fires) and runs it once the trigger is back, a caller on a route held on the line; hibernate does the same for `graceMinutes` with its questions hidden, then takes nothing more; wipe drops it all."
},
{
"name": "running",
Expand Down
13 changes: 13 additions & 0 deletions catalog/instances/lifecycle.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@
//! choices with the same defaults.

use weft::instance::InstanceId;
use weft::node::NodeOutput;
use weft::program::InfraDownAnswer;
use weft::{node_bail, DeactivateSpec, DeactivationMode, ExecutionContext, RunningPolicy, StopSelf, WeftResult};

/// The instance id a node was handed, checked the way every instance id
Expand Down Expand Up @@ -46,3 +48,14 @@ pub fn take_down_spec(ctx: &ExecutionContext) -> WeftResult<DeactivateSpec> {
pub fn stop_self(ctx: &ExecutionContext) -> WeftResult<StopSelf> {
Ok(if ctx.inputs.get::<bool>("includeSelf")? { StopSelf::Include } else { StopSelf::Keep })
}

/// What a node that takes one instance's copy down sends on: `done` when
/// the copy is down (taken down now, or already down), `noCopy` when that
/// instance has no copy of the node (a mistyped id, or one terminated
/// before), so a program answering a person can tell them which.
pub fn taken_down(answer: InfraDownAnswer) -> NodeOutput {
match answer {
InfraDownAnswer::TakenDown | InfraDownAnswer::AlreadyDown => NodeOutput::new().set("done", true),
InfraDownAnswer::NoCopy => NodeOutput::new().set("noCopy", true),
}
}
9 changes: 7 additions & 2 deletions catalog/instances/stop_instance_infra/metadata.json
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@
},
"default": "park",
"label": "Triggers",
"description": "The instance's triggers reading it: park queues what arrives, hibernate keeps accepting for `graceMinutes`, wipe drops it."
"description": "The instance's triggers reading it: park keeps everything that arrives (calls, answers, the trigger's own fires) and runs it once the trigger is back, a caller on a route held on the line; hibernate does the same for `graceMinutes` with its questions hidden, then takes nothing more; wipe drops it all. Starting the copy again brings back the ones it took down."
},
{
"name": "running",
Expand Down Expand Up @@ -75,7 +75,12 @@
{
"name": "done",
"type": "Boolean",
"description": "True once the triggers reading the copy are down and the stop is handed off; the copy scales down afterwards."
"description": "True once the triggers reading the copy are down and the stop is handed off (or the copy was already down); the copy scales down afterwards."
},
{
"name": "noCopy",
"type": "Boolean",
"description": "True instead of `done` when this instance has no copy of the node: the id names no instance that has one, or its copy was terminated before. Wire it to tell whoever asked."
}
],
"validate": [
Expand Down
7 changes: 3 additions & 4 deletions catalog/instances/stop_instance_infra/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,9 @@

use async_trait::async_trait;

use weft::node::NodeOutput;
use weft::{ExecutionContext, Node, NodeManifest, WeftResult};

use super::lifecycle::{instance, stop_self, take_down_spec};
use super::lifecycle::{instance, stop_self, take_down_spec, taken_down};

#[derive(NodeManifest)]
pub struct StopInstanceInfraNode;
Expand All @@ -25,7 +24,7 @@ impl Node for StopInstanceInfraNode {
async fn run(&self, ctx: ExecutionContext) -> WeftResult<()> {
let node: String = ctx.inputs.get("node")?;
let spec = take_down_spec(&ctx)?;
ctx.infra(node).instance(instance(&ctx)?).stop(spec, stop_self(&ctx)?).await?;
ctx.pulse_downstream(NodeOutput::new().set("done", true)).await
let answer = ctx.infra(node).instance(instance(&ctx)?).stop(spec, stop_self(&ctx)?).await?;
ctx.pulse_downstream(taken_down(answer)).await
}
}
18 changes: 18 additions & 0 deletions catalog/instances/stop_instance_infra/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,27 @@ pub fn tests() -> Vec<NodeTest> {
NodeTest::fake("parks_and_waits_by_default", defaults),
NodeTest::fake("the_choices_reach_the_call", choices),
NodeTest::fake("waiting_before_a_wipe_is_refused", wipe_wait),
NodeTest::fake("no_copy_is_told_apart_from_done", no_copy),
NodeTest::fake("a_copy_already_down_is_done", already_down),
]
}

async fn no_copy(rig: FakeRig) -> WeftResult<()> {
rig.answer_program_call("weft.infra.stop", json!({ "copy": "no_copy" }));
let out = rig.run(&StopInstanceInfraNode, json!({ "node": "bridge", "instance": "adaa" })).await.ok()?;
assert_eq!(out.output("noCopy")?, &json!(true));
assert!(out.output("done").is_err(), "nothing was taken down");
Ok(())
}

async fn already_down(rig: FakeRig) -> WeftResult<()> {
rig.answer_program_call("weft.infra.stop", json!({ "copy": "already_down" }));
let out = rig.run(&StopInstanceInfraNode, json!({ "node": "bridge", "instance": "ada" })).await.ok()?;
assert_eq!(out.output("done")?, &json!(true));
assert!(out.output("noCopy").is_err());
Ok(())
}

async fn defaults(rig: FakeRig) -> WeftResult<()> {
rig.run(&StopInstanceInfraNode, json!({ "node": "bridge", "instance": "ada" })).await.ok()?;
match &rig.program_calls()[0] {
Expand Down
9 changes: 7 additions & 2 deletions catalog/instances/terminate_instance_infra/metadata.json
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@
},
"default": "park",
"label": "Triggers",
"description": "The instance's triggers reading it: park queues what arrives, hibernate keeps accepting for `graceMinutes`, wipe drops it."
"description": "The instance's triggers reading it: park keeps everything that arrives (calls, answers, the trigger's own fires) and runs it once the trigger is back, a caller on a route held on the line; hibernate does the same for `graceMinutes` with its questions hidden, then takes nothing more; wipe drops it all."
},
{
"name": "running",
Expand Down Expand Up @@ -76,7 +76,12 @@
{
"name": "done",
"type": "Boolean",
"description": "True once the triggers reading the copy are down and the delete is handed off; the copy is deleted afterwards."
"description": "True once the triggers reading the copy are down and the delete is handed off (or the copy was already going); the copy is deleted afterwards."
},
{
"name": "noCopy",
"type": "Boolean",
"description": "True instead of `done` when this instance has no copy of the node: the id names no instance that has one, or its copy was terminated before. Wire it to tell whoever asked."
}
],
"validate": [
Expand Down
7 changes: 3 additions & 4 deletions catalog/instances/terminate_instance_infra/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,9 @@

use async_trait::async_trait;

use weft::node::NodeOutput;
use weft::{ExecutionContext, Node, NodeManifest, WeftResult};

use super::lifecycle::{instance, stop_self, take_down_spec};
use super::lifecycle::{instance, stop_self, take_down_spec, taken_down};

#[derive(NodeManifest)]
pub struct TerminateInstanceInfraNode;
Expand All @@ -25,7 +24,7 @@ impl Node for TerminateInstanceInfraNode {
async fn run(&self, ctx: ExecutionContext) -> WeftResult<()> {
let node: String = ctx.inputs.get("node")?;
let spec = take_down_spec(&ctx)?;
ctx.infra(node).instance(instance(&ctx)?).terminate(spec, stop_self(&ctx)?).await?;
ctx.pulse_downstream(NodeOutput::new().set("done", true)).await
let answer = ctx.infra(node).instance(instance(&ctx)?).terminate(spec, stop_self(&ctx)?).await?;
ctx.pulse_downstream(taken_down(answer)).await
}
}
82 changes: 72 additions & 10 deletions crates/weft-broker-client/src/activation.rs
Original file line number Diff line number Diff line change
Expand Up @@ -29,10 +29,13 @@ pub struct ActivationLifecycle {
/// second past which the drain gives up, cancels what is still running,
/// and lands Inactive (enforced by the stuck-transition reaper).
pub drain_deadline_unix: Option<i64>,
/// True iff the CURRENT deactivation was the health loop parking this
/// activation because the infra it reads broke. Its auto-recover
/// reactivates only such an activation, never one a person took down.
pub deactivated_by_health: bool,
/// Why the CURRENT deactivation happened, when it was the infra the
/// trigger reads going down rather than a person: the health loop
/// parking it because a copy broke, or an infra verb (stop, terminate,
/// upgrade) taking it down with its copies. The copy coming back
/// brings it back (the health loop's auto-recover, an infra start);
/// one a person took down stays down until they switch it on.
pub went_down: Option<WentDown>,
/// The trigger-setup run of the activation in flight, reserved before
/// setup is queued; every write the activation makes is guarded by it.
/// `None` outside Activating.
Expand All @@ -53,7 +56,7 @@ impl ActivationLifecycle {
fires_visible_to_consumers: true,
fires_deadline_unix: None,
drain_deadline_unix: None,
deactivated_by_health: false,
went_down: None,
activating_execution_id: None,
}
}
Expand All @@ -80,8 +83,8 @@ impl ActivationLifecycle {
}
}

/// Hibernate: parking until the deadline, then refusing; hidden the
/// whole time.
/// Hibernate: parking until the deadline, then refusing (the dispatcher
/// clears `accepting_fires` once it passes); hidden the whole time.
pub fn hibernating(deadline_unix: i64) -> Self {
Self {
status: ProjectStatus::Inactive,
Expand All @@ -107,6 +110,20 @@ impl ActivationLifecycle {
}
}

/// Whether this is a wipe: inactive, refusing work, with no grace
/// window (a hibernation whose window ended refuses work too, and keeps
/// its deadline).
// SYNC: is_wiped <-> WIPE_FORGETS_SQL (crates/weft-dispatcher/src/activation_store.rs)
pub fn is_wiped(&self) -> bool {
self.status == ProjectStatus::Inactive && !self.accepting_fires && self.fires_deadline_unix.is_none()
}

/// Whether the health loop took this down and it is still down: what
/// its auto-recover brings back.
pub fn parked_by_health(&self) -> bool {
self.went_down.is_some_and(|w| w.with == DownWith::Health) && self.status == ProjectStatus::Inactive
}

/// Where the activation stands, as a person reads it: `registered`,
/// `activating`, `active`, `deactivating`, or for an inactive
/// activation how it went down: `wipe`, `hibernate`, `park`.
Expand All @@ -118,7 +135,7 @@ impl ActivationLifecycle {
ProjectStatus::Activating => ActivationMode::Activating,
ProjectStatus::Active => ActivationMode::Active,
ProjectStatus::Deactivating => ActivationMode::Deactivating,
ProjectStatus::Inactive => ActivationMode::Down(if !self.accepting_fires {
ProjectStatus::Inactive => ActivationMode::Down(if self.is_wiped() {
DeactivationMode::Wipe
} else if !self.fires_visible_to_consumers {
DeactivationMode::Hibernate
Expand Down Expand Up @@ -151,7 +168,7 @@ pub fn aggregate<'a>(lifecycles: impl IntoIterator<Item = &'a ActivationLifecycl
fires_visible_to_consumers: draining.iter().any(|l| l.fires_visible_to_consumers),
fires_deadline_unix: draining.iter().filter_map(|l| l.fires_deadline_unix).max(),
drain_deadline_unix: draining.iter().filter_map(|l| l.drain_deadline_unix).max(),
deactivated_by_health: draining.iter().all(|l| l.deactivated_by_health),
went_down: same_cause(&draining),
activating_execution_id: None,
};
}
Expand All @@ -168,11 +185,56 @@ pub fn aggregate<'a>(lifecycles: impl IntoIterator<Item = &'a ActivationLifecycl
fires_visible_to_consumers: inactive.iter().any(|l| l.fires_visible_to_consumers),
fires_deadline_unix: inactive.iter().filter_map(|l| l.fires_deadline_unix).max(),
drain_deadline_unix: None,
deactivated_by_health: inactive.iter().all(|l| l.deactivated_by_health),
went_down: same_cause(&inactive),
activating_execution_id: None,
}
}

/// The cause every one of `lifecycles` went down with, when they all went
/// down with the same one, at the latest of their moments.
fn same_cause(lifecycles: &[&ActivationLifecycle]) -> Option<WentDown> {
let with = lifecycles.first()?.went_down?.with;
let mut at_unix = i64::MIN;
for lifecycle in lifecycles {
let went = lifecycle.went_down.filter(|w| w.with == with)?;
at_unix = at_unix.max(went.at_unix);
}
Some(WentDown { with, at_unix })
}

/// What took a trigger down when it was not a person, and when (see
/// [`ActivationLifecycle::went_down`]). The moment is what tells a copy
/// that came back up since from one still going down: only the first
/// brings the trigger back.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WentDown {
pub with: DownWith,
pub at_unix: i64,
}

/// What took a trigger down when it was not a person ([`WentDown`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DownWith {
/// The health loop parked it: a copy it reads broke.
Health,
/// An infra verb took it down with the copies it reads.
Infra,
}

impl DownWith {
// SYNC: the spellings <-> trigger_activation.went_down_with (crates/weft-dispatcher/src/activation_store.rs)
pub fn as_str(self) -> &'static str {
match self {
DownWith::Health => "health",
DownWith::Infra => "infra",
}
}

pub fn parse(s: &str) -> Option<Self> {
[DownWith::Health, DownWith::Infra].into_iter().find(|cause| cause.as_str() == s)
}
}

#[cfg(test)]
mod tests {
use super::*;
Expand Down
Loading
Loading