broker: upgrade and rollback keep a stopped actor stopped
A stopped actor gets the new version installed, keeps its mailbox and does not run main until the next start. Running actors still restart on the new version. The upgraded and rolled_back answers say whether the actor runs.
This commit is contained in:
+4
-3
@@ -125,8 +125,8 @@ the request: the request's `type` in past tense (`spawned`, `upgraded`,
|
|||||||
| `type` | Fields | Effect |
|
| `type` | Fields | Effect |
|
||||||
| --- | --- | --- |
|
| --- | --- | --- |
|
||||||
| `spawn` | `actor`, `wasm`, `source`?, `built_by`? | test, then start a new actor |
|
| `spawn` | `actor`, `wasm`, `source`?, `built_by`? | test, then start a new actor |
|
||||||
| `upgrade` | `actor`, `wasm`, `source`?, `built_by`? | test, then swap the version |
|
| `upgrade` | `actor`, `wasm`, `source`?, `built_by`? | test, then swap the version: a running actor is stopped (normal shutdown) and starts the new version; a stopped actor gets the new version and stays stopped, with its mailbox untouched, until `start`. The answer has `running` |
|
||||||
| `rollback` | `actor`, `version` | go back to an older version |
|
| `rollback` | `actor`, `version` | go back to an older version; like `upgrade`, a stopped actor stays stopped. The answer has `running` |
|
||||||
| `stop`, `start` | `actor` | stop or start an actor |
|
| `stop`, `start` | `actor` | stop or start an actor |
|
||||||
| `remove` | `actor` | stop the actor (normal shutdown), then delete its whole folder `/actor/<id>/` (all versions, `data/`, `public/`, logs); answer `{"type": "removed", "actor"}`. Messages to it are then undeliverable, and the ID can be spawned again. An unknown actor or a native one (`broker`, `console`, `cargo`, `fs`) is an `error` |
|
| `remove` | `actor` | stop the actor (normal shutdown), then delete its whole folder `/actor/<id>/` (all versions, `data/`, `public/`, logs); answer `{"type": "removed", "actor"}`. Messages to it are then undeliverable, and the ID can be spawned again. An unknown actor or a native one (`broker`, `console`, `cargo`, `fs`) is an `error` |
|
||||||
| `subscribe`, `unsubscribe` | `actor` | receive (or not) what `actor` publishes |
|
| `subscribe`, `unsubscribe` | `actor` | receive (or not) what `actor` publishes |
|
||||||
@@ -212,7 +212,8 @@ instance. A version without test files passes.
|
|||||||
|
|
||||||
**Swap.** The broker holds the mailbox, stops the old instance, makes the
|
**Swap.** The broker holds the mailbox, stops the old instance, makes the
|
||||||
top-level `actor.wasm` and `source/` a copy of the new version, and starts
|
top-level `actor.wasm` and `source/` a copy of the new version, and starts
|
||||||
it. A rollback does the same with the older version.
|
it. A stopped actor is not started: it keeps its mailbox and runs the new
|
||||||
|
version at the next `start`. A rollback does the same with the older version.
|
||||||
|
|
||||||
**Crashes.** The broker restarts a crashed actor after 0.5 s, then doubles
|
**Crashes.** The broker restarts a crashed actor after 0.5 s, then doubles
|
||||||
the delay up to 30 s. A new version that crashes 3 times within 30 seconds
|
the delay up to 30 s. A new version that crashes 3 times within 30 seconds
|
||||||
|
|||||||
+5
-4
@@ -331,12 +331,13 @@ impl Broker {
|
|||||||
Ok((component, tests))
|
Ok((component, tests))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Make `version` the current `actor.wasm`, `source/`, and `actor.json`.
|
/// Make `version` the current `actor.wasm`, `source/`, and `actor.json`,
|
||||||
pub fn install(&self, id: &str, version: u64) -> Result<()> {
|
/// and record whether the actor runs.
|
||||||
|
pub fn install(&self, id: &str, version: u64, running: bool) -> Result<()> {
|
||||||
self.world.install_files(id, version)?;
|
self.world.install_files(id, version)?;
|
||||||
self.update_meta(id, |m| {
|
self.update_meta(id, |m| {
|
||||||
m.version = Some(version);
|
m.version = Some(version);
|
||||||
m.running = true;
|
m.running = running;
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -360,7 +361,7 @@ impl Broker {
|
|||||||
|
|
||||||
async fn create_version(self: &Arc<Self>, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<Value> {
|
async fn create_version(self: &Arc<Self>, id: &str, meta: &VersionMeta, wasm: &Path, source: Option<&Path>) -> Result<Value> {
|
||||||
let (_, tests) = self.try_version(id, meta, wasm, source).await?;
|
let (_, tests) = self.try_version(id, meta, wasm, source).await?;
|
||||||
self.install(id, meta.version)?;
|
self.install(id, meta.version, true)?;
|
||||||
self.start_supervisor(id, meta.version, true);
|
self.start_supervisor(id, meta.version, true);
|
||||||
Ok(json!({"type": "spawned", "actor": id, "version": meta.version, "tests": tests}))
|
Ok(json!({"type": "spawned", "actor": id, "version": meta.version, "tests": tests}))
|
||||||
}
|
}
|
||||||
|
|||||||
+10
-5
@@ -220,17 +220,16 @@ impl Supervisor {
|
|||||||
|
|
||||||
/// Make `version` current. It is compiled first (unless `component` is
|
/// Make `version` current. It is compiled first (unless `component` is
|
||||||
/// given): if that fails, nothing changes. `data/` and `public/` stay as
|
/// given): if that fails, nothing changes. `data/` and `public/` stay as
|
||||||
/// they are.
|
/// they are. A stopped actor stays stopped.
|
||||||
async fn install(&mut self, version: u64, component: Option<Component>) -> Result<()> {
|
async fn install(&mut self, version: u64, component: Option<Component>) -> Result<()> {
|
||||||
let component = match component {
|
let component = match component {
|
||||||
Some(c) => c,
|
Some(c) => c,
|
||||||
None => self.broker.load_version(&self.id, version)?,
|
None => self.broker.load_version(&self.id, version)?,
|
||||||
};
|
};
|
||||||
self.stop().await;
|
self.stop().await;
|
||||||
self.broker.install(&self.id, version)?;
|
self.broker.install(&self.id, version, self.running)?;
|
||||||
self.version = version;
|
self.version = version;
|
||||||
self.component = Some(component);
|
self.component = Some(component);
|
||||||
self.running = true;
|
|
||||||
self.restarts = 0;
|
self.restarts = 0;
|
||||||
self.restart_at = None;
|
self.restart_at = None;
|
||||||
Ok(())
|
Ok(())
|
||||||
@@ -239,20 +238,26 @@ impl Supervisor {
|
|||||||
async fn upgrade(&mut self, version: u64, component: Component, requester: String) -> Result<Value> {
|
async fn upgrade(&mut self, version: u64, component: Component, requester: String) -> Result<Value> {
|
||||||
let previous = self.version;
|
let previous = self.version;
|
||||||
self.install(version, Some(component)).await?;
|
self.install(version, Some(component)).await?;
|
||||||
|
if !self.running {
|
||||||
|
// Installed, not started: main runs and mail is read at the next `start`.
|
||||||
|
return Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous, "running": false}));
|
||||||
|
}
|
||||||
self.probation = Some(Probation { previous, since: Instant::now(), crashes: 0, requester });
|
self.probation = Some(Probation { previous, since: Instant::now(), crashes: 0, requester });
|
||||||
if let Err(e) = self.start().await {
|
if let Err(e) = self.start().await {
|
||||||
self.rollback(previous).await?;
|
self.rollback(previous).await?;
|
||||||
bail!("version {version} did not start ({e:#}); rolled back to {previous}");
|
bail!("version {version} did not start ({e:#}); rolled back to {previous}");
|
||||||
}
|
}
|
||||||
Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous}))
|
Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous, "running": true}))
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn rollback(&mut self, version: u64) -> Result<Value> {
|
async fn rollback(&mut self, version: u64) -> Result<Value> {
|
||||||
let from = self.version;
|
let from = self.version;
|
||||||
self.install(version, None).await?;
|
self.install(version, None).await?;
|
||||||
self.probation = None;
|
self.probation = None;
|
||||||
|
if self.running {
|
||||||
self.start().await?;
|
self.start().await?;
|
||||||
Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": version}))
|
}
|
||||||
|
Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": version, "running": self.running}))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Handle a control request. Returns `true` when the supervisor ends.
|
/// Handle a control request. Returns `true` when the supervisor ends.
|
||||||
|
|||||||
@@ -506,6 +506,50 @@ async fn rolls_back_a_crash_loop_to_the_previous_version_without_restoring_data(
|
|||||||
host.stop().await;
|
host.stop().await;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn upgrade_and_rollback_keep_a_stopped_actor_stopped() {
|
||||||
|
let world = create_world("upgrade-stopped");
|
||||||
|
let wasm = toolbox_wasm().clone();
|
||||||
|
let (host, mut mailbox) = start(&world).await;
|
||||||
|
let source = passing_source(&world, "source");
|
||||||
|
spawn_type(&host, &mut mailbox, "sleeper", &wasm, &source).await;
|
||||||
|
let stopped = request(&host, &mut mailbox, json!({"type": "stop", "actor": "sleeper"})).await;
|
||||||
|
assert_eq!(stopped.body["type"], "stopped");
|
||||||
|
|
||||||
|
let upgraded = upgrade_actor(&host, &mut mailbox, "sleeper", &wasm, &source).await;
|
||||||
|
assert_eq!(upgraded.body["type"], "upgraded", "{upgraded:?}");
|
||||||
|
assert_eq!(upgraded.body["version"], 2);
|
||||||
|
assert_eq!(upgraded.body["running"], false);
|
||||||
|
let meta = world.read_meta("sleeper");
|
||||||
|
assert_eq!((meta.version, meta.running), (Some(2), false));
|
||||||
|
|
||||||
|
// Mail waits in the mailbox: the stopped actor reads nothing.
|
||||||
|
let waiting = host.broker.send("console", Some("sleeper"), None, None, json!({"type": "version"}));
|
||||||
|
assert!(
|
||||||
|
tokio::time::timeout(Duration::from_millis(500), receive_reply(&mut mailbox, &waiting, Duration::from_secs(10)))
|
||||||
|
.await
|
||||||
|
.is_err(),
|
||||||
|
"a stopped actor must not answer"
|
||||||
|
);
|
||||||
|
|
||||||
|
let rolled_back = request(&host, &mut mailbox, json!({"type": "rollback", "actor": "sleeper", "version": 1})).await;
|
||||||
|
assert_eq!(rolled_back.body["type"], "rolled_back", "{rolled_back:?}");
|
||||||
|
assert_eq!(rolled_back.body["running"], false);
|
||||||
|
assert!(!world.read_meta("sleeper").running);
|
||||||
|
|
||||||
|
// `start` runs the installed version and delivers the waiting message.
|
||||||
|
let started = request(&host, &mut mailbox, json!({"type": "start", "actor": "sleeper"})).await;
|
||||||
|
assert_eq!(started.body["type"], "started");
|
||||||
|
let answer = receive_reply(&mut mailbox, &waiting, Duration::from_secs(10)).await;
|
||||||
|
assert_eq!(answer.body["version"], "1");
|
||||||
|
|
||||||
|
// A running actor still restarts on the new version.
|
||||||
|
let running = upgrade_actor(&host, &mut mailbox, "sleeper", &wasm, &source).await;
|
||||||
|
assert_eq!(running.body["running"], true);
|
||||||
|
assert_eq!(ask(&host, &mut mailbox, "sleeper", json!({"type": "version"})).await.body["version"], "3");
|
||||||
|
host.stop().await;
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn explicit_rollback_keeps_current_data() {
|
async fn explicit_rollback_keeps_current_data() {
|
||||||
let world = create_world("explicit-rollback");
|
let world = create_world("explicit-rollback");
|
||||||
|
|||||||
Reference in New Issue
Block a user