mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: enforce final provider upgrade heartbeat

Browse Source
This commit is contained in:
Codex
2026-05-24 05:39:48 +00:00
parent ce0b6db548
commit d148bbadd2
6 changed files with 766 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/backend-core/src/provider_registry.rs
+649 -2
View File
@@ -3,12 +3,13 @@ use std::sync::Arc;
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
use axum::http::{HeaderMap, Uri};
use axum::response::Response;
use chrono::{DateTime, Duration, Utc};
use futures_util::{SinkExt, StreamExt};
use serde_json::{json, Value};
use tokio::sync::mpsc;
use crate::db::{
record_event, update_provider_heartbeat, upsert_docker_status, upsert_provider_node,
raw_task, record_event, update_provider_heartbeat, upsert_docker_status, upsert_provider_node,
upsert_system_status,
};
use crate::egress_tcp::{handle_egress_tcp_close, handle_egress_tcp_data, handle_egress_tcp_open};
@@ -17,6 +18,513 @@ use crate::json_util::compact_json_for_storage;
use crate::ssh_bridge::forward_ssh_provider_message;
use crate::state::{AppState, HttpTunnelResponse, ProviderConnection};
use crate::task_dispatcher::{is_terminal_task_status, notify_task_terminal};
use crate::types::RawTaskRow;
const PROVIDER_UPGRADE_FINAL_HEARTBEAT_TIMEOUT_MS: i64 = 240_000;
const PROVIDER_UPGRADE_FINAL_HEARTBEAT_POLL_MS: u64 = 2_000;
#[derive(Clone, Debug)]
struct ProviderUpgradeFinalHeartbeatExpectation {
task_id: String,
provider_id: String,
target_version: String,
min_heartbeat_at: DateTime<Utc>,
deadline_at: DateTime<Utc>,
scheduler_result: Value,
candidate_name: Option<String>,
last_known_good_version: Option<String>,
}
#[derive(Clone, Debug)]
struct ProviderUpgradeNodeSnapshot {
provider_id: String,
status: String,
labels: Value,
}
#[derive(Clone, Debug)]
enum ProviderUpgradeFinalHeartbeatDecision {
Pending(Value),
Succeeded(Value),
Failed(Value),
}
fn json_string_at(value: &Value, path: &[&str]) -> Option<String> {
let mut current = value;
for part in path {
current = current.get(*part)?;
}
current
.as_str()
.filter(|text| !text.is_empty())
.map(ToOwned::to_owned)
}
fn provider_upgrade_label_string(labels: &Value, key: &str) -> Option<String> {
labels
.get(key)
.and_then(Value::as_str)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
fn provider_upgrade_heartbeat_at(snapshot: &ProviderUpgradeNodeSnapshot) -> Option<DateTime<Utc>> {
provider_upgrade_label_string(&snapshot.labels, "providerGatewayHeartbeatAt").and_then(
|value| {
DateTime::parse_from_rfc3339(&value)
.ok()
.map(|parsed| parsed.with_timezone(&Utc))
},
)
}
fn provider_upgrade_schedule_task(task: &RawTaskRow) -> bool {
task.command == "provider.upgrade"
&& task.payload.get("mode").and_then(Value::as_str) == Some("schedule")
}
fn provider_upgrade_target_version(task: &RawTaskRow, scheduler_result: &Value) -> Option<String> {
json_string_at(scheduler_result, &["targetProviderGatewayVersion"])
.or_else(|| json_string_at(scheduler_result, &["plan", "targetProviderGatewayVersion"]))
.or_else(|| json_string_at(&task.payload, &["targetProviderGatewayVersion"]))
}
fn provider_upgrade_last_known_good_version(scheduler_result: &Value) -> Option<String> {
json_string_at(scheduler_result, &["providerGatewayVersion"])
.or_else(|| json_string_at(scheduler_result, &["plan", "providerGatewayVersion"]))
}
fn provider_upgrade_candidate_name(scheduler_result: &Value) -> Option<String> {
json_string_at(scheduler_result, &["candidateName"])
.or_else(|| json_string_at(scheduler_result, &["plan", "candidateName"]))
}
fn provider_upgrade_expectation(
task: &RawTaskRow,
scheduler_result: &Value,
now: DateTime<Utc>,
) -> Option<ProviderUpgradeFinalHeartbeatExpectation> {
let target_version = provider_upgrade_target_version(task, scheduler_result)?;
Some(ProviderUpgradeFinalHeartbeatExpectation {
task_id: task.id.clone(),
provider_id: task.provider_id.clone(),
target_version,
min_heartbeat_at: now,
deadline_at: now + Duration::milliseconds(PROVIDER_UPGRADE_FINAL_HEARTBEAT_TIMEOUT_MS),
scheduler_result: scheduler_result.clone(),
candidate_name: provider_upgrade_candidate_name(scheduler_result),
last_known_good_version: provider_upgrade_last_known_good_version(scheduler_result),
})
}
fn provider_upgrade_snapshot_is_candidate(
snapshot: &ProviderUpgradeNodeSnapshot,
candidate_name: Option<&str>,
) -> bool {
let container_name =
provider_upgrade_label_string(&snapshot.labels, "providerGatewayContainerName");
if let (Some(container_name), Some(candidate_name)) =
(container_name.as_deref(), candidate_name)
{
if container_name == candidate_name {
return true;
}
}
container_name
.as_deref()
.is_some_and(|name| name.contains("-candidate-"))
}
fn insert_json_field_with_unavailable_reason(
object: &mut serde_json::Map<String, Value>,
key: &str,
value: Option<String>,
unavailable_reason: &str,
) {
if let Some(value) = value {
object.insert(key.to_string(), Value::String(value));
} else {
object.insert(key.to_string(), Value::Null);
object.insert(
format!("{key}UnavailableReason"),
Value::String(unavailable_reason.to_string()),
);
}
}
fn provider_upgrade_final_container_json(
snapshot: Option<&ProviderUpgradeNodeSnapshot>,
unavailable_reason: &str,
) -> Value {
let labels = snapshot.map(|snapshot| &snapshot.labels);
let heartbeat = snapshot
.and_then(provider_upgrade_heartbeat_at)
.map(|value| value.to_rfc3339_opts(chrono::SecondsFormat::Millis, true));
let mut object = serde_json::Map::new();
insert_json_field_with_unavailable_reason(
&mut object,
"containerId",
labels
.and_then(|labels| provider_upgrade_label_string(labels, "providerGatewayContainerId")),
unavailable_reason,
);
insert_json_field_with_unavailable_reason(
&mut object,
"containerName",
labels.and_then(|labels| {
provider_upgrade_label_string(labels, "providerGatewayContainerName")
}),
unavailable_reason,
);
insert_json_field_with_unavailable_reason(
&mut object,
"version",
labels.and_then(|labels| provider_upgrade_label_string(labels, "providerGatewayVersion")),
unavailable_reason,
);
insert_json_field_with_unavailable_reason(
&mut object,
"restartPolicy",
labels.and_then(|labels| {
provider_upgrade_label_string(labels, "providerGatewayRestartPolicy")
}),
unavailable_reason,
);
insert_json_field_with_unavailable_reason(
&mut object,
"pidMode",
labels.and_then(|labels| provider_upgrade_label_string(labels, "providerGatewayPidMode")),
unavailable_reason,
);
insert_json_field_with_unavailable_reason(
&mut object,
"heartbeatTimestamp",
heartbeat,
unavailable_reason,
);
Value::Object(object)
}
fn evaluate_provider_upgrade_final_heartbeat(
expectation: &ProviderUpgradeFinalHeartbeatExpectation,
snapshot: Option<&ProviderUpgradeNodeSnapshot>,
now: DateTime<Utc>,
) -> ProviderUpgradeFinalHeartbeatDecision {
let candidate_snapshot = snapshot.is_some_and(|snapshot| {
provider_upgrade_snapshot_is_candidate(snapshot, expectation.candidate_name.as_deref())
});
let unavailable_reason = if candidate_snapshot {
"latest heartbeat came from candidate validation container, not final promoted compose service"
} else {
"final promoted compose heartbeat not observed"
};
let final_container_snapshot = if candidate_snapshot { None } else { snapshot };
if let Some(snapshot) = snapshot {
let observed_version =
provider_upgrade_label_string(&snapshot.labels, "providerGatewayVersion");
let heartbeat_is_new = provider_upgrade_heartbeat_at(snapshot)
.is_some_and(|last| last >= expectation.min_heartbeat_at);
let same_provider = snapshot.provider_id.as_str() == expectation.provider_id.as_str();
let online = snapshot.status == "online";
if same_provider
&& online
&& observed_version.as_deref() == Some(expectation.target_version.as_str())
&& heartbeat_is_new
&& !candidate_snapshot
{
let restart_policy =
provider_upgrade_label_string(&snapshot.labels, "providerGatewayRestartPolicy");
let pid_mode =
provider_upgrade_label_string(&snapshot.labels, "providerGatewayPidMode");
if restart_policy.as_deref() != Some("always") || pid_mode.as_deref() != Some("host") {
return ProviderUpgradeFinalHeartbeatDecision::Failed(json!({
"ok": false,
"reason": "final-promoted-runtime-guard-failed",
"blocker": true,
"providerId": expectation.provider_id.as_str(),
"taskId": expectation.task_id.as_str(),
"targetProviderGatewayVersion": expectation.target_version.as_str(),
"finalPromotedContainer": provider_upgrade_final_container_json(Some(snapshot), "runtime guard heartbeat fields are incomplete"),
"rollback": {
"preferred": "last-known-good",
"required": true,
"lastKnownGoodProviderGatewayVersion": expectation.last_known_good_version.as_deref(),
"updaterRollbackExpected": true
},
"candidateValidationWasTerminal": false,
"schedulerResult": expectation.scheduler_result.clone()
}));
}
return ProviderUpgradeFinalHeartbeatDecision::Succeeded(json!({
"ok": true,
"reason": "final-promoted-heartbeat-confirmed",
"providerId": expectation.provider_id.as_str(),
"taskId": expectation.task_id.as_str(),
"targetProviderGatewayVersion": expectation.target_version.as_str(),
"finalPromotedContainer": provider_upgrade_final_container_json(Some(snapshot), "not reported by provider heartbeat"),
"heartbeat": {
"minHeartbeatAt": expectation.min_heartbeat_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
"observedAt": provider_upgrade_heartbeat_at(snapshot).map(|value| value.to_rfc3339_opts(chrono::SecondsFormat::Millis, true))
},
"candidateValidationWasTerminal": false,
"schedulerResult": expectation.scheduler_result.clone()
}));
}
}
let pending_result = json!({
"ok": false,
"status": "running",
"reason": "awaiting-final-promoted-heartbeat",
"providerId": expectation.provider_id.as_str(),
"taskId": expectation.task_id.as_str(),
"targetProviderGatewayVersion": expectation.target_version.as_str(),
"deadlineAt": expectation.deadline_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
"candidateValidationWasTerminal": false,
"finalPromotedContainer": provider_upgrade_final_container_json(final_container_snapshot, unavailable_reason),
"schedulerResult": expectation.scheduler_result.clone()
});
if now < expectation.deadline_at {
return ProviderUpgradeFinalHeartbeatDecision::Pending(pending_result);
}
ProviderUpgradeFinalHeartbeatDecision::Failed(json!({
"ok": false,
"reason": "final-promoted-heartbeat-timeout",
"error": "final promoted provider-gateway compose container did not report a target-version heartbeat before the guard window expired",
"blocker": true,
"providerId": expectation.provider_id.as_str(),
"taskId": expectation.task_id.as_str(),
"targetProviderGatewayVersion": expectation.target_version.as_str(),
"heartbeat": {
"minHeartbeatAt": expectation.min_heartbeat_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
"deadlineAt": expectation.deadline_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
"observedAt": snapshot.and_then(provider_upgrade_heartbeat_at).map(|value| value.to_rfc3339_opts(chrono::SecondsFormat::Millis, true))
},
"finalPromotedContainer": provider_upgrade_final_container_json(final_container_snapshot, unavailable_reason),
"rollback": {
"preferred": "last-known-good",
"required": true,
"lastKnownGoodProviderGatewayVersion": expectation.last_known_good_version.as_deref(),
"updaterRollbackExpected": true
},
"candidateValidationWasTerminal": false,
"schedulerResult": expectation.scheduler_result.clone()
}))
}
async fn provider_upgrade_node_snapshot(
state: &Arc<AppState>,
provider_id: &str,
) -> anyhow::Result<Option<ProviderUpgradeNodeSnapshot>> {
if !state.db_ready() {
return Ok(None);
}
let client = state.pool.get().await?;
let row = client
.query_opt(
"SELECT provider_id, status, labels FROM unidesk_nodes WHERE provider_id = $1 LIMIT 1",
&[&provider_id],
)
.await?;
Ok(row.map(|row| ProviderUpgradeNodeSnapshot {
provider_id: row.get("provider_id"),
status: row.get("status"),
labels: row.get("labels"),
}))
}
async fn complete_provider_upgrade_task(
state: &Arc<AppState>,
task_id: &str,
provider_id: &str,
status: &str,
result: Value,
) -> anyhow::Result<()> {
let client = state.pool.get().await?;
let updated = client
.execute(
r#"
UPDATE unidesk_tasks
SET status = $2, result = $3::jsonb, updated_at = now()
WHERE id = $1
AND command = 'provider.upgrade'
AND status NOT IN ('succeeded', 'failed')
"#,
&[&task_id, &status, &result],
)
.await?;
record_event(
state,
"provider_upgrade_final_heartbeat_result",
provider_id,
json!({
"providerId": provider_id,
"taskId": task_id,
"status": status,
"result": result,
"updated": updated,
}),
)
.await;
if updated > 0 && is_terminal_task_status(status) {
notify_task_terminal(state, task_id).await?;
}
Ok(())
}
async fn mark_provider_upgrade_awaiting_final_heartbeat(
state: &Arc<AppState>,
expectation: &ProviderUpgradeFinalHeartbeatExpectation,
pending_result: Value,
) -> anyhow::Result<()> {
let client = state.pool.get().await?;
client
.execute(
r#"
UPDATE unidesk_tasks
SET status = 'running', result = $2::jsonb, updated_at = now()
WHERE id = $1
AND command = 'provider.upgrade'
AND status NOT IN ('succeeded', 'failed')
"#,
&[&expectation.task_id, &pending_result],
)
.await?;
record_event(
state,
"provider_upgrade_final_heartbeat_wait",
&expectation.provider_id,
json!({
"providerId": expectation.provider_id.as_str(),
"taskId": expectation.task_id.as_str(),
"targetProviderGatewayVersion": expectation.target_version.as_str(),
"deadlineAt": expectation.deadline_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
"candidateValidationWasTerminal": false,
}),
)
.await;
Ok(())
}
async fn monitor_provider_upgrade_final_heartbeat(
state: Arc<AppState>,
expectation: ProviderUpgradeFinalHeartbeatExpectation,
) -> anyhow::Result<()> {
loop {
let snapshot = provider_upgrade_node_snapshot(&state, &expectation.provider_id).await?;
match evaluate_provider_upgrade_final_heartbeat(&expectation, snapshot.as_ref(), Utc::now())
{
ProviderUpgradeFinalHeartbeatDecision::Succeeded(result) => {
complete_provider_upgrade_task(
&state,
&expectation.task_id,
&expectation.provider_id,
"succeeded",
result,
)
.await?;
break;
}
ProviderUpgradeFinalHeartbeatDecision::Failed(result) => {
complete_provider_upgrade_task(
&state,
&expectation.task_id,
&expectation.provider_id,
"failed",
result,
)
.await?;
break;
}
ProviderUpgradeFinalHeartbeatDecision::Pending(_) => {
tokio::time::sleep(std::time::Duration::from_millis(
PROVIDER_UPGRADE_FINAL_HEARTBEAT_POLL_MS,
))
.await;
}
}
}
Ok(())
}
async fn ensure_provider_upgrade_final_heartbeat_watcher(
state: &Arc<AppState>,
expectation: ProviderUpgradeFinalHeartbeatExpectation,
) {
let task_id = expectation.task_id.clone();
let inserted = {
let mut active = state.active_provider_upgrade_watchers.lock().await;
active.insert(task_id.clone())
};
if !inserted {
return;
}
let state_for_task = state.clone();
tokio::spawn(async move {
let provider_id = expectation.provider_id.clone();
if let Err(error) =
monitor_provider_upgrade_final_heartbeat(state_for_task.clone(), expectation).await
{
state_for_task.log(
"error",
"provider_upgrade_final_heartbeat_watcher_failed",
Some(json!({ "taskId": task_id.as_str(), "providerId": provider_id.as_str(), "error": error.to_string() })),
);
}
let mut active = state_for_task.active_provider_upgrade_watchers.lock().await;
active.remove(&task_id);
});
}
async fn maybe_handle_provider_upgrade_schedule_status(
state: &Arc<AppState>,
provider_id: &str,
task_id: &str,
status: &str,
stored_result: &Value,
) -> anyhow::Result<bool> {
let starts_final_guard = status == "succeeded"
|| (status == "running"
&& stored_result.get("mode").and_then(Value::as_str) == Some("schedule")
&& stored_result
.get("terminalStatusManagedByBackendCore")
.and_then(Value::as_bool)
== Some(true));
if !starts_final_guard {
return Ok(false);
}
let Some(task) = raw_task(state, task_id).await? else {
return Ok(false);
};
if !provider_upgrade_schedule_task(&task) {
return Ok(false);
}
let now = Utc::now();
let Some(expectation) = provider_upgrade_expectation(&task, stored_result, now) else {
let result = json!({
"ok": false,
"reason": "target-provider-gateway-version-unavailable",
"error": "provider.upgrade mode=schedule cannot be finalized because targetProviderGatewayVersion was not reported",
"blocker": true,
"providerId": provider_id,
"taskId": task_id,
"finalPromotedContainer": provider_upgrade_final_container_json(None, "target gateway version unavailable before final heartbeat guard"),
"candidateValidationWasTerminal": false,
"schedulerResult": stored_result
});
complete_provider_upgrade_task(state, task_id, provider_id, "failed", result).await?;
return Ok(true);
};
let pending = match evaluate_provider_upgrade_final_heartbeat(&expectation, None, now) {
ProviderUpgradeFinalHeartbeatDecision::Pending(result) => result,
ProviderUpgradeFinalHeartbeatDecision::Succeeded(result)
| ProviderUpgradeFinalHeartbeatDecision::Failed(result) => result,
};
mark_provider_upgrade_awaiting_final_heartbeat(state, &expectation, pending).await?;
ensure_provider_upgrade_final_heartbeat_watcher(state, expectation).await;
Ok(true)
}
pub async fn provider_ws(
state: Arc<AppState>,
@@ -202,7 +710,16 @@ async fn handle_provider_text(
let _ = connection.sender.send(Message::Text(json!({ "type": "ack", "requestId": "register", "ok": true, "message": "registered" }).to_string()));
}
"heartbeat" => {
let labels = message.get("labels").cloned().unwrap_or_else(|| json!({}));
let mut labels = message.get("labels").cloned().unwrap_or_else(|| json!({}));
if let Some(object) = labels.as_object_mut() {
object.insert(
"providerGatewayHeartbeatAt".to_string(),
message
.get("at")
.cloned()
.unwrap_or_else(|| Value::String(Utc::now().to_rfc3339())),
);
}
update_provider_heartbeat(state, &provider_id, &labels).await?;
state.log(
"debug",
@@ -241,6 +758,17 @@ async fn handle_provider_text(
compact_json_for_storage(&message.get("result").cloned().unwrap_or_else(
|| json!({ "message": message.get("message").cloned().unwrap_or(Value::Null) }),
));
if maybe_handle_provider_upgrade_schedule_status(
state,
&provider_id,
task_id,
status,
&stored_result,
)
.await?
{
return Ok(());
}
let client = state.pool.get().await?;
client.execute(
r#"
@@ -339,3 +867,122 @@ pub async fn mark_stale_providers_offline(state: &Arc<AppState>) -> anyhow::Resu
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn dt(value: &str) -> DateTime<Utc> {
DateTime::parse_from_rfc3339(value)
.expect("valid fixture timestamp")
.with_timezone(&Utc)
}
fn expectation() -> ProviderUpgradeFinalHeartbeatExpectation {
ProviderUpgradeFinalHeartbeatExpectation {
task_id: "task_upgrade_contract".to_string(),
provider_id: "D601".to_string(),
target_version: "0.2.27".to_string(),
min_heartbeat_at: dt("2026-05-24T10:00:00Z"),
deadline_at: dt("2026-05-24T10:04:00Z"),
scheduler_result: json!({
"mode": "schedule",
"providerGatewayVersion": "0.2.26",
"targetProviderGatewayVersion": "0.2.27",
"plan": {
"candidateName": "unidesk-provider-gateway-D601-candidate-task-upgrade-contract"
}
}),
candidate_name: Some(
"unidesk-provider-gateway-D601-candidate-task-upgrade-contract".to_string(),
),
last_known_good_version: Some("0.2.26".to_string()),
}
}
#[test]
fn candidate_validation_heartbeat_is_not_terminal_success() {
let snapshot = ProviderUpgradeNodeSnapshot {
provider_id: "D601".to_string(),
status: "online".to_string(),
labels: json!({
"providerGatewayVersion": "0.2.27",
"providerGatewayContainerId": "candidate-container-id",
"providerGatewayContainerName": "unidesk-provider-gateway-D601-candidate-task-upgrade-contract",
"providerGatewayRestartPolicy": "no",
"providerGatewayPidMode": "host",
"providerGatewayHeartbeatAt": "2026-05-24T10:01:00Z"
}),
};
let decision = evaluate_provider_upgrade_final_heartbeat(
&expectation(),
Some(&snapshot),
dt("2026-05-24T10:05:00Z"),
);
let ProviderUpgradeFinalHeartbeatDecision::Failed(result) = decision else {
panic!("candidate heartbeat must fail after final promoted heartbeat timeout");
};
assert_eq!(
result.get("reason").and_then(Value::as_str),
Some("final-promoted-heartbeat-timeout")
);
assert_eq!(
result
.get("candidateValidationWasTerminal")
.and_then(Value::as_bool),
Some(false)
);
assert_eq!(
result
.pointer("/rollback/preferred")
.and_then(Value::as_str),
Some("last-known-good")
);
assert_eq!(
result
.pointer("/finalPromotedContainer/containerIdUnavailableReason")
.and_then(Value::as_str),
Some("latest heartbeat came from candidate validation container, not final promoted compose service")
);
}
#[test]
fn final_promoted_compose_heartbeat_is_success() {
let snapshot = ProviderUpgradeNodeSnapshot {
provider_id: "D601".to_string(),
status: "online".to_string(),
labels: json!({
"providerGatewayVersion": "0.2.27",
"providerGatewayContainerId": "final-container-id",
"providerGatewayContainerName": "unidesk-provider-gateway-D601",
"providerGatewayRestartPolicy": "always",
"providerGatewayPidMode": "host",
"providerGatewayHeartbeatAt": "2026-05-24T10:02:00Z"
}),
};
let decision = evaluate_provider_upgrade_final_heartbeat(
&expectation(),
Some(&snapshot),
dt("2026-05-24T10:02:01Z"),
);
let ProviderUpgradeFinalHeartbeatDecision::Succeeded(result) = decision else {
panic!("final compose heartbeat should succeed");
};
assert_eq!(
result.get("reason").and_then(Value::as_str),
Some("final-promoted-heartbeat-confirmed")
);
assert_eq!(
result
.pointer("/finalPromotedContainer/containerId")
.and_then(Value::as_str),
Some("final-container-id")
);
assert_eq!(
result
.pointer("/finalPromotedContainer/heartbeatTimestamp")
.and_then(Value::as_str),
Some("2026-05-24T10:02:00.000Z")
);
}
}
src/components/backend-core/src/state.rs
+3 -1
View File
@@ -1,4 +1,4 @@
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use std::str::FromStr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
@@ -51,6 +51,7 @@ pub struct AppState {
pub request_performance_samples: Mutex<Vec<RequestPerformanceSample>>,
pub operation_performance_samples: Mutex<Vec<OperationPerformanceSample>>,
pub active_scheduled_runs: Mutex<std::collections::HashSet<String>>,
pub active_provider_upgrade_watchers: Mutex<HashSet<String>>,
}
impl AppState {
@@ -84,6 +85,7 @@ impl AppState {
request_performance_samples: Mutex::new(Vec::new()),
operation_performance_samples: Mutex::new(Vec::new()),
active_scheduled_runs: Mutex::new(std::collections::HashSet::new()),
active_provider_upgrade_watchers: Mutex::new(HashSet::new()),
})
}
src/components/microservices/code-queue/Dockerfile
+6 -1
View File
@@ -3,8 +3,11 @@ FROM ${CODE_QUEUE_BASE_IMAGE}
ENV PLAYWRIGHT_BROWSERS_PATH=/ms-playwright
ENV UNIDESK_SKILLS_PATH=/root/.agents/skills
ENV RUSTUP_HOME=/usr/local/rustup
ENV CARGO_HOME=/usr/local/cargo
ENV PATH=/usr/local/cargo/bin:${PATH}
RUN (command -v codex >/dev/null 2>&1 && command -v opencode >/dev/null 2>&1 && command -v docker >/dev/null 2>&1 && command -v gh >/dev/null 2>&1 && command -v rg >/dev/null 2>&1 && command -v cargo >/dev/null 2>&1 && command -v rustc >/dev/null 2>&1 && command -v rustfmt >/dev/null 2>&1 && command -v xvfb-run >/dev/null 2>&1 && command -v xauth >/dev/null 2>&1 && test -x "$PLAYWRIGHT_BROWSERS_PATH/chromium_headless_shell-1217/chrome-headless-shell-linux64/chrome-headless-shell") \
RUN (command -v codex >/dev/null 2>&1 && command -v opencode >/dev/null 2>&1 && command -v docker >/dev/null 2>&1 && command -v gh >/dev/null 2>&1 && command -v rg >/dev/null 2>&1 && command -v cargo >/dev/null 2>&1 && command -v rustc >/dev/null 2>&1 && rustc --version | awk '{ split($2, v, "."); exit ! (v[1] > 1 || (v[1] == 1 && v[2] >= 86)) }' && command -v rustfmt >/dev/null 2>&1 && command -v xvfb-run >/dev/null 2>&1 && command -v xauth >/dev/null 2>&1 && test -x "$PLAYWRIGHT_BROWSERS_PATH/chromium_headless_shell-1217/chrome-headless-shell-linux64/chrome-headless-shell") \
|| (apt-get update \
&& apt-get install -y --no-install-recommends \
bash \
@@ -42,6 +45,8 @@ RUN (command -v codex >/dev/null 2>&1 && command -v opencode >/dev/null 2>&1 &&
xauth \
xvfb \
xz-utils \
&& curl https://sh.rustup.rs -sSf | sh -s -- -y --profile minimal --default-toolchain stable \
&& rustup component add rustfmt \
&& mkdir -p /usr/local/lib/docker/cli-plugins /root/.docker/cli-plugins \
&& ln -sf /usr/bin/docker-compose /usr/local/lib/docker/cli-plugins/docker-compose \
&& ln -sf /usr/bin/docker-compose /root/.docker/cli-plugins/docker-compose \
src/components/provider-gateway/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@unidesk/provider-gateway",
"version": "0.2.26",
"version": "0.2.27",
"private": true,
"type": "module",
"scripts": {
src/components/provider-gateway/src/index.ts
+101 -13
View File
@@ -1671,7 +1671,24 @@ function upgradePlan(taskId: string): Record<string, JsonValue> {
const validationNeedleAck = `"requestId":"register"`;
const validationNeedleOk = `"ok":true`;
const validationAttempts = Math.max(1, Math.ceil(validationTimeoutMs / 2000));
const finalPromotedReconnectTimeoutMs = 180_000;
const finalPromotedHeartbeatTimeoutMs = 240_000;
const finalValidationAttempts = Math.max(1, Math.ceil(finalPromotedReconnectTimeoutMs / 2000));
const targetGatewayMetadata = readTargetGatewayMetadata(workspace);
const unavailableFinalPromotedContainer = {
containerId: null,
containerIdUnavailableReason: "mode=plan or schedule has not yet observed backend-core final promoted heartbeat",
containerName: null,
containerNameUnavailableReason: "mode=plan or schedule has not yet observed backend-core final promoted heartbeat",
version: null,
versionUnavailableReason: "mode=plan or schedule has not yet observed backend-core final promoted heartbeat",
restartPolicy: null,
restartPolicyUnavailableReason: "mode=plan or schedule has not yet observed backend-core final promoted heartbeat",
pidMode: null,
pidModeUnavailableReason: "mode=plan or schedule has not yet observed backend-core final promoted heartbeat",
heartbeatTimestamp: null,
heartbeatTimestampUnavailableReason: "mode=plan or schedule has not yet observed backend-core final promoted heartbeat",
};
const compactBuildContextRefreshScript = [
`provider_src=${shellQuote(`${workspaceRoot}/src/components/provider-gateway`)}`,
`shared_src=${shellQuote(`${workspaceRoot}/src/components/shared`)}`,
@@ -1693,6 +1710,33 @@ function upgradePlan(taskId: string): Record<string, JsonValue> {
` fi`,
`done`,
].join("\n");
const rollbackToLastKnownGoodScript = [
`rollback_to_last_known_good() {`,
` reason="$1"`,
` echo "rollback_to_last_known_good: $reason" >&2`,
` if [ -n "$final_container" ]; then docker rm -f "$final_container" >/dev/null 2>&1 || true; fi`,
` docker rm -f ${shellQuote(candidateName)} >/dev/null 2>&1 || true`,
` if [ -z "$old_name" ] || [ -z "$old_image" ]; then echo "rollback_to_last_known_good unavailable: old_name=$old_name old_image=$old_image" >&2; return 1; fi`,
` docker rm -f "$old_name" >/dev/null 2>&1 || true`,
[
` docker run -d`,
`--name "$old_name"`,
`--restart always`,
`--pid "\${old_pid_mode:-host}"`,
`$network_arg`,
`--env-file "$candidate_env_file"`,
`--label ${shellQuote(`com.docker.compose.project=${config.upgradeComposeProject}`)}`,
`--label ${shellQuote(`com.docker.compose.service=${config.upgradeService}`)}`,
`--label ${shellQuote("com.docker.compose.oneoff=False")}`,
`--label ${shellQuote(`unidesk.upgrade.rollback=${taskId}`)}`,
`$mount_args`,
`$extra_host_args`,
`"$old_image" >/dev/null || return 1`,
].join(" "),
` if [ -n "$network_names" ]; then for net in $network_names; do if [ "$net" != "$first_network" ]; then docker network connect "$net" "$old_name" >/dev/null 2>&1 || true; fi; done; fi`,
` echo "rollback_to_last_known_good restored old gateway container name=$old_name image=$old_image reason=$reason" >&2`,
`}`,
].join("\n");
const script = [
"set -eu",
`cd ${shellQuote(workspace)}`,
@@ -1701,16 +1745,24 @@ function upgradePlan(taskId: string): Record<string, JsonValue> {
`first_old=""`,
`for old_id in $old_ids; do first_old="$old_id"; break; done`,
`old_name=""`,
`old_image=""`,
`old_restart=""`,
`old_pid_mode=""`,
`final_container=""`,
`mount_args=""`,
`extra_host_args=""`,
`candidate_env_file=${shellQuote(`/tmp/${candidateName}.env`)}`,
`network_names=""`,
`first_network=""`,
`network_arg=""`,
rollbackToLastKnownGoodScript,
`if [ -n "$old_ids" ]; then docker update --restart always $old_ids >/dev/null 2>&1 || true; fi`,
`if [ -z "$first_old" ]; then echo "no existing provider-gateway compose container found; cannot perform safe in-place upgrade" >&2; exit 1; fi`,
`old_name=$(docker inspect --format '{{.Name}}' "$first_old")`,
`old_name="\${old_name#/}"`,
`old_image=$(docker inspect --format '{{.Image}}' "$first_old")`,
`old_restart=$(docker inspect --format '{{.HostConfig.RestartPolicy.Name}}' "$first_old")`,
`old_pid_mode=$(docker inspect --format '{{.HostConfig.PidMode}}' "$first_old")`,
`docker inspect --format '{{range .Config.Env}}{{println .}}{{end}}' "$first_old" > "$candidate_env_file"`,
`mount_args=$(docker inspect --format '{{range .Mounts}}{{printf "-v %s:%s" .Source .Destination}}{{if not .RW}}{{printf ":ro"}}{{end}}{{printf "\\n"}}{{end}}' "$first_old")`,
`extra_host_args=$(docker inspect --format '{{range .HostConfig.ExtraHosts}}{{printf "--add-host %s\\n" .}}{{end}}' "$first_old")`,
@@ -1745,12 +1797,18 @@ function upgradePlan(taskId: string): Record<string, JsonValue> {
`docker rm -f ${shellQuote(candidateName)} >/dev/null 2>&1 || true`,
composeUpCommand.map(shellQuote).join(" "),
`final_container="$old_name"`,
`if [ -z "$final_container" ]; then final_container=$(${listServiceContainersCommand.map(shellQuote).join(" ")} | head -n 1); fi`,
`if [ -z "$final_container" ] || ! docker inspect "$final_container" >/dev/null 2>&1; then final_container=$(${listServiceContainersCommand.map(shellQuote).join(" ")} | head -n 1); fi`,
`if [ -z "$final_container" ]; then echo "final provider-gateway compose container not found after promotion" >&2; rollback_to_last_known_good "final-compose-container-missing" || true; rm -f "$candidate_env_file"; exit 1; fi`,
`final_container_id=$(docker inspect --format '{{.Id}}' "$final_container")`,
`final_restart=$(docker inspect --format '{{.HostConfig.RestartPolicy.Name}}' "$final_container")`,
`final_pid_mode=$(docker inspect --format '{{.HostConfig.PidMode}}' "$final_container")`,
`if [ "$final_restart" != "always" ] || [ "$final_pid_mode" != "host" ]; then echo "final provider-gateway runtime guard failed: restart=$final_restart pid=$final_pid_mode" >&2; exit 1; fi`,
`if [ "$final_restart" != "always" ] || [ "$final_pid_mode" != "host" ]; then echo "final provider-gateway runtime guard failed: restart=$final_restart pid=$final_pid_mode" >&2; rollback_to_last_known_good "final-runtime-guard-failed" || true; rm -f "$candidate_env_file"; exit 1; fi`,
`final_attempt=0`,
`final_validated=0`,
`while [ "$final_attempt" -lt "${finalValidationAttempts}" ]; do final_logs=$(docker logs "$final_container" 2>&1 || true); final_has_open=0; final_has_ack=0; final_has_ok=0; case "$final_logs" in *${shellQuote(validationNeedleOpen)}*) final_has_open=1;; esac; case "$final_logs" in *${shellQuote(validationNeedleAck)}*) final_has_ack=1;; esac; case "$final_logs" in *${shellQuote(validationNeedleOk)}*) final_has_ok=1;; esac; if [ "$final_has_open" = "1" ] && [ "$final_has_ack" = "1" ] && [ "$final_has_ok" = "1" ]; then final_validated=1; break; fi; final_running=$(docker inspect --format '{{.State.Running}}' "$final_container" 2>/dev/null || true); if [ "$final_running" != "true" ]; then break; fi; final_attempt=$((final_attempt + 1)); sleep 2; done`,
`if [ "$final_validated" != "1" ]; then echo "final provider-gateway compose container did not reconnect to backend-core before local guard timeout; backend-core must not treat candidate validation as success" >&2; docker logs "$final_container" >&2 || true; rollback_to_last_known_good "final-promoted-container-reconnect-timeout" || true; rm -f "$candidate_env_file"; exit 1; fi`,
`rm -f "$candidate_env_file"`,
`echo "candidate provider-gateway validated and promoted"`,
`echo "candidate provider-gateway validated; final promoted compose container reconnected locally: container=$final_container id=$final_container_id restart=$final_restart pid=$final_pid_mode"`,
].join("\n");
const dockerRunCommand = [
"docker",
@@ -1805,16 +1863,23 @@ function upgradePlan(taskId: string): Record<string, JsonValue> {
buildBeforeCandidate: true,
oldGatewaySleepMs: sleepMs,
validationTimeoutMs,
finalPromotedReconnectTimeoutMs,
finalPromotedHeartbeatTimeoutMs,
oldGatewayRestartPolicyBeforeSleep: "always",
promoteOnlyAfterCandidateValidation: true,
candidateRestartPolicyAfterPromotion: "always",
candidateFinalRestartPolicyValidation: true,
candidateUsesOldContainerMounts: true,
candidateUsesOldContainerNetworks: true,
candidateUsesOldContainerExtraHosts: true,
candidateUsesOldContainerEnvironment: true,
candidateUsesHostPidNamespace: true,
startupSelfHealsRestartPolicy: true,
candidateValidationIsNotTerminalSuccess: true,
finalPromotedComposeHeartbeatRequired: true,
finalPromotedContainerMustNotBeCandidate: true,
localRollbackToLastKnownGoodOnFinalReconnectFailure: true,
backendCoreOwnsFinalTerminalStatus: true,
candidateRestartPolicyAfterPromotion: "always",
candidateFinalRestartPolicyValidation: true,
candidateUsesOldContainerMounts: true,
candidateUsesOldContainerNetworks: true,
candidateUsesOldContainerExtraHosts: true,
candidateUsesOldContainerEnvironment: true,
candidateUsesHostPidNamespace: true,
startupSelfHealsRestartPolicy: true,
dockerStatusReportsRestartPolicyAndPidMode: true,
refreshesCompactBuildContextBeforeBuild: true,
removeScope: {
@@ -1825,6 +1890,17 @@ function upgradePlan(taskId: string): Record<string, JsonValue> {
namedVolumesPreserved: true,
},
dockerRunCommand,
finalPromotedHeartbeatConfirmation: {
required: true,
providerId: config.providerId,
targetProviderGatewayVersion: targetGatewayMetadata.version,
backendCoreNodeRegistry: "/api/nodes",
heartbeatTimeoutMs: finalPromotedHeartbeatTimeoutMs,
finalContainerMustNotEqualCandidate: candidateName,
requiredResultFields: ["containerId", "version", "restartPolicy", "pidMode", "heartbeatTimestamp"],
unavailableResult: unavailableFinalPromotedContainer,
},
finalPromotedContainer: unavailableFinalPromotedContainer,
compactBuildContextCandidates,
workspaceStateMount: `${config.upgradeHostProjectRoot.replace(/\/+$/, "")}/.state:${workspaceRoot}/.state`,
};
@@ -1843,12 +1919,16 @@ async function runProviderUpgrade(taskId: string, payload: Record<string, JsonVa
setTimeout(() => enterUpgradeSleep(sleepMs), 250).unref?.();
return {
mode,
message: "provider gateway upgrade scheduled with sleep-and-validate rollback guard",
message: "provider gateway upgrade scheduled; backend-core must wait for final promoted compose heartbeat before terminal success",
providerId: config.providerId,
providerName: config.providerName,
providerGatewayVersion: gatewayMetadata.version,
targetProviderGatewayVersion: (plan.targetProviderGatewayVersion as string | undefined) ?? gatewayMetadata.version,
updaterContainerId: result.stdout.trim(),
terminalStatusManagedByBackendCore: true,
candidateValidationIsNotTerminalSuccess: true,
finalPromotedHeartbeatConfirmation: plan.finalPromotedHeartbeatConfirmation,
finalPromotedContainer: plan.finalPromotedContainer,
plan,
stderr: result.stderr.slice(0, 500),
};
@@ -2145,7 +2225,15 @@ async function handleDispatch(message: CoreDispatchMessage): Promise<void> {
}
if (message.command === "provider.upgrade") {
const result = await runProviderUpgrade(message.taskId, message.payload);
await sendTaskStatus(message.taskId, "succeeded", "provider upgrade command completed", result);
const scheduleMode = message.payload.mode === "schedule";
await sendTaskStatus(
message.taskId,
scheduleMode ? "running" : "succeeded",
scheduleMode
? "provider upgrade scheduled; awaiting backend-core final promoted heartbeat"
: "provider upgrade command completed",
result,
);
return;
}
if (message.command === "host.ssh") {