mirror of
https://github.com/farion1231/cc-switch.git
synced 2026-08-04 11:43:57 +08:00
fix(pi): bind recovery to durable native authority
This commit is contained in:
@@ -20,6 +20,7 @@ use std::sync::{Arc, LazyLock, Mutex, MutexGuard};
|
||||
|
||||
use super::shared_file::{
|
||||
compare_exchange_shared_file_bytes, delete_shared_file, read_shared_file,
|
||||
sync_shared_file_parent,
|
||||
};
|
||||
|
||||
const MAX_PI_MODELS_BYTES: u64 = 8 * 1024 * 1024;
|
||||
@@ -35,6 +36,9 @@ static PI_JSON_TRAILING_COMMAS: LazyLock<Regex> = LazyLock::new(|| {
|
||||
});
|
||||
static PATH_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Arc<Mutex<()>>>>> =
|
||||
LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||
#[cfg(test)]
|
||||
static BEFORE_PROVIDER_VERIFY: LazyLock<Mutex<HashMap<PathBuf, Vec<u8>>>> =
|
||||
LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||
|
||||
fn path_lock(path: &Path) -> Result<Arc<Mutex<()>>, AppError> {
|
||||
let mut locks = PATH_LOCKS
|
||||
@@ -379,23 +383,7 @@ fn apply_pi_provider_patch_checked(
|
||||
for _ in 0..MAX_MUTATION_ATTEMPTS {
|
||||
let before = read_models_bytes(path)?;
|
||||
if let Some(expected) = expected {
|
||||
let observed = provider_values_from_bytes(path, before.as_deref(), expected.keys())?;
|
||||
if let Some((provider_key, expected_value)) =
|
||||
expected.iter().find(|(provider_key, expected_value)| {
|
||||
observed.get(*provider_key) != Some(*expected_value)
|
||||
})
|
||||
{
|
||||
return Err(AppError::Conflict(format!(
|
||||
"Pi provider key '{provider_key}' changed since directory/catalog preflight \
|
||||
(expected {}, observed {})",
|
||||
provider_value_label(expected_value),
|
||||
provider_value_label(
|
||||
observed
|
||||
.get(provider_key)
|
||||
.expect("every requested provider key is observed")
|
||||
)
|
||||
)));
|
||||
}
|
||||
ensure_provider_values_match(path, before.as_deref(), expected)?;
|
||||
}
|
||||
if before.is_none() && patch.values().all(Option::is_none) {
|
||||
return Ok(false);
|
||||
@@ -460,6 +448,30 @@ fn provider_values_from_bytes<'a>(
|
||||
.collect())
|
||||
}
|
||||
|
||||
fn ensure_provider_values_match(
|
||||
path: &Path,
|
||||
bytes: Option<&[u8]>,
|
||||
expected: &IndexMap<String, Option<Value>>,
|
||||
) -> Result<(), AppError> {
|
||||
let observed = provider_values_from_bytes(path, bytes, expected.keys())?;
|
||||
if let Some((provider_key, expected_value)) = expected
|
||||
.iter()
|
||||
.find(|(provider_key, expected_value)| observed.get(*provider_key) != Some(*expected_value))
|
||||
{
|
||||
return Err(AppError::Conflict(format!(
|
||||
"Pi provider key '{provider_key}' changed since directory/catalog preflight \
|
||||
(expected {}, observed {})",
|
||||
provider_value_label(expected_value),
|
||||
provider_value_label(
|
||||
observed
|
||||
.get(provider_key)
|
||||
.expect("every requested provider key is observed")
|
||||
)
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn provider_value_label(value: &Option<Value>) -> &'static str {
|
||||
if value.is_some() {
|
||||
"present"
|
||||
@@ -499,7 +511,24 @@ impl PiProviderPatchReceipt {
|
||||
/// concurrent edit of an owned key turns compensation into an explicit
|
||||
/// conflict instead of being overwritten.
|
||||
pub(crate) fn rollback(&self) -> Result<(), AppError> {
|
||||
apply_pi_provider_patch_checked(&self.path, Some(&self.attempted), &self.before.values)?;
|
||||
if let Err(error) =
|
||||
apply_pi_provider_patch_checked(&self.path, Some(&self.attempted), &self.before.values)
|
||||
{
|
||||
let observed =
|
||||
snapshot_pi_provider_values(&self.path, self.before.values.keys().cloned())?;
|
||||
if observed.values != self.before.values {
|
||||
return Err(error);
|
||||
}
|
||||
// A namespace mutation may have committed before its durability
|
||||
// barrier reported failure. Re-observe the exact semantic state
|
||||
// and retry the parent sync before declaring compensation done.
|
||||
sync_shared_file_parent(&self.path).map_err(|sync_error| {
|
||||
AppError::Config(format!(
|
||||
"Pi provider rollback reached the previous exact-key state but could not \
|
||||
confirm directory durability ({error}; retry={sync_error})"
|
||||
))
|
||||
})?;
|
||||
}
|
||||
remove_new_empty_document(&self.path, &self.before)
|
||||
}
|
||||
}
|
||||
@@ -520,6 +549,103 @@ pub(crate) fn snapshot_pi_provider_values(
|
||||
})
|
||||
}
|
||||
|
||||
/// Revalidate a previously captured exact-key set without mutating the
|
||||
/// document. This is the ownership-claim barrier used when runtime takeover is
|
||||
/// disabled and therefore has no gateway projection write of its own.
|
||||
pub(crate) fn verify_pi_provider_values(
|
||||
path: &Path,
|
||||
expected: &IndexMap<String, Option<Value>>,
|
||||
) -> Result<(), AppError> {
|
||||
let lock = path_lock(path)?;
|
||||
let _guard = lock_path(&lock)?;
|
||||
#[cfg(test)]
|
||||
if let Some(replacement) = BEFORE_PROVIDER_VERIFY
|
||||
.lock()
|
||||
.map_err(|error| AppError::Lock(error.to_string()))?
|
||||
.remove(path)
|
||||
{
|
||||
fs::write(path, replacement).map_err(|error| AppError::io(path, error))?;
|
||||
}
|
||||
let bytes = read_models_bytes(path)?;
|
||||
ensure_provider_values_match(path, bytes.as_deref(), expected)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn replace_before_next_pi_provider_verify(path: &Path, bytes: &[u8]) {
|
||||
BEFORE_PROVIDER_VERIFY
|
||||
.lock()
|
||||
.expect("Pi provider verify hook lock")
|
||||
.insert(path.to_path_buf(), bytes.to_vec());
|
||||
}
|
||||
|
||||
fn attempted_provider_values(
|
||||
before: &PiProviderValuesSnapshot,
|
||||
patch: &IndexMap<String, Option<Value>>,
|
||||
) -> IndexMap<String, Option<Value>> {
|
||||
before
|
||||
.values
|
||||
.iter()
|
||||
.map(|(provider_key, previous)| {
|
||||
(
|
||||
provider_key.clone(),
|
||||
patch
|
||||
.get(provider_key)
|
||||
.cloned()
|
||||
.unwrap_or_else(|| previous.clone()),
|
||||
)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn failed_publish_after_conditional_compensation(
|
||||
path: &Path,
|
||||
before: &PiProviderValuesSnapshot,
|
||||
attempted: &IndexMap<String, Option<Value>>,
|
||||
publish_error: AppError,
|
||||
) -> AppError {
|
||||
if matches!(&publish_error, AppError::Conflict(_)) {
|
||||
return publish_error;
|
||||
}
|
||||
let observed = match snapshot_pi_provider_values(path, before.values.keys().cloned()) {
|
||||
Ok(observed) => observed,
|
||||
Err(observe_error) => {
|
||||
return AppError::Config(format!(
|
||||
"Pi provider publication failed ({publish_error}) and its commit state could not \
|
||||
be inspected for compensation: {observe_error}"
|
||||
));
|
||||
}
|
||||
};
|
||||
if observed.values == before.values {
|
||||
if let Err(sync_error) = sync_shared_file_parent(path) {
|
||||
return AppError::Config(format!(
|
||||
"Pi provider publication failed ({publish_error}); the previous exact-key state \
|
||||
is visible but directory durability could not be confirmed: {sync_error}"
|
||||
));
|
||||
}
|
||||
return publish_error;
|
||||
}
|
||||
if observed.values != *attempted {
|
||||
return AppError::Config(format!(
|
||||
"Pi provider publication failed ({publish_error}); the live exact-key state was \
|
||||
superseded before conditional compensation and was preserved"
|
||||
));
|
||||
}
|
||||
|
||||
let receipt = PiProviderPatchReceipt {
|
||||
path: path.to_path_buf(),
|
||||
before: before.clone(),
|
||||
attempted: attempted.clone(),
|
||||
attempted_file_existed: observed.file_existed,
|
||||
};
|
||||
match receipt.rollback() {
|
||||
Ok(()) => publish_error,
|
||||
Err(rollback_error) => AppError::Config(format!(
|
||||
"Pi provider publication failed ({publish_error}) after the attempted exact-key state \
|
||||
became live; conditional compensation also failed: {rollback_error}"
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Publish an exact-key patch only if every preflighted provider value still
|
||||
/// matches. Whole-file CAS retries preserve unrelated edits, but re-check the
|
||||
/// provider precondition before every retry.
|
||||
@@ -536,20 +662,15 @@ pub(crate) fn apply_pi_provider_patch_with_receipt(
|
||||
"Pi provider patch contains a key which was not preflighted".to_string(),
|
||||
));
|
||||
}
|
||||
let changed = apply_pi_provider_patch_checked(path, Some(&before.values), patch)?;
|
||||
let attempted = before
|
||||
.values
|
||||
.iter()
|
||||
.map(|(provider_key, previous)| {
|
||||
(
|
||||
provider_key.clone(),
|
||||
patch
|
||||
.get(provider_key)
|
||||
.cloned()
|
||||
.unwrap_or_else(|| previous.clone()),
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
let attempted = attempted_provider_values(before, patch);
|
||||
let changed = match apply_pi_provider_patch_checked(path, Some(&before.values), patch) {
|
||||
Ok(changed) => changed,
|
||||
Err(error) => {
|
||||
return Err(failed_publish_after_conditional_compensation(
|
||||
path, before, &attempted, error,
|
||||
));
|
||||
}
|
||||
};
|
||||
Ok(PiProviderPatchReceipt {
|
||||
path: path.to_path_buf(),
|
||||
before: before.clone(),
|
||||
@@ -569,17 +690,37 @@ fn remove_new_empty_document(
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let current = read_shared_file(path, MAX_PI_MODELS_BYTES, "Pi models file")?;
|
||||
let canonical_empty = serialize_models_mutation(path, None, &|_| Ok(()))?;
|
||||
if current.bytes.as_deref() == Some(canonical_empty.as_slice()) {
|
||||
delete_shared_file(
|
||||
let mut last_error = None;
|
||||
for _ in 0..MAX_MUTATION_ATTEMPTS {
|
||||
let current = read_shared_file(path, MAX_PI_MODELS_BYTES, "Pi models file")?;
|
||||
match current.bytes.as_deref() {
|
||||
None => {
|
||||
if last_error.is_some() {
|
||||
sync_shared_file_parent(path)?;
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
Some(bytes) if bytes != canonical_empty.as_slice() => {
|
||||
// A concurrent writer added unrelated content after our key
|
||||
// rollback. File ownership therefore belongs to that writer.
|
||||
return Ok(());
|
||||
}
|
||||
Some(_) => {}
|
||||
}
|
||||
match delete_shared_file(
|
||||
path,
|
||||
¤t.revision,
|
||||
MAX_PI_MODELS_BYTES,
|
||||
"Pi models file",
|
||||
)?;
|
||||
) {
|
||||
Ok(_) => return Ok(()),
|
||||
Err(error) => last_error = Some(error),
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
Err(last_error.unwrap_or_else(|| {
|
||||
AppError::Config("Pi empty models document cleanup did not make progress".to_string())
|
||||
}))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
@@ -755,6 +896,86 @@ mod tests {
|
||||
assert_eq!(document["providers"]["managed"]["api"], "external");
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn create_parent_sync_failure_conditionally_removes_the_committed_provider() {
|
||||
let temp = tempfile::tempdir().expect("tempdir");
|
||||
let path = temp.path().join("models.json");
|
||||
let before =
|
||||
snapshot_pi_provider_values(&path, ["managed".to_string()]).expect("preflight");
|
||||
crate::pi_config::shared_file::fail_next_parent_sync_for_test(&path);
|
||||
|
||||
let error = apply_pi_provider_patch_with_receipt(
|
||||
&path,
|
||||
&before,
|
||||
&IndexMap::from([(
|
||||
"managed".to_string(),
|
||||
Some(serde_json::json!({"api": "attempted"})),
|
||||
)]),
|
||||
)
|
||||
.expect_err("durability failure must remain visible");
|
||||
assert!(error.to_string().contains("injected"));
|
||||
assert!(
|
||||
!path.exists(),
|
||||
"a failed create must not leave a committed provider or empty shadow file"
|
||||
);
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn replace_parent_sync_failure_conditionally_restores_the_previous_provider() {
|
||||
let temp = tempfile::tempdir().expect("tempdir");
|
||||
let path = temp.path().join("models.json");
|
||||
fs::write(
|
||||
&path,
|
||||
r#"{"root":"preserved","providers":{"managed":{"api":"before"}}}"#,
|
||||
)
|
||||
.expect("seed");
|
||||
let before =
|
||||
snapshot_pi_provider_values(&path, ["managed".to_string()]).expect("preflight");
|
||||
crate::pi_config::shared_file::fail_next_parent_sync_for_test(&path);
|
||||
|
||||
apply_pi_provider_patch_with_receipt(
|
||||
&path,
|
||||
&before,
|
||||
&IndexMap::from([(
|
||||
"managed".to_string(),
|
||||
Some(serde_json::json!({"api": "attempted"})),
|
||||
)]),
|
||||
)
|
||||
.expect_err("durability failure must remain visible");
|
||||
let restored: Value =
|
||||
serde_json::from_slice(&fs::read(&path).expect("restored document")).expect("parse");
|
||||
assert_eq!(restored["root"], "preserved");
|
||||
assert_eq!(restored["providers"]["managed"]["api"], "before");
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn remove_key_parent_sync_failure_conditionally_restores_the_deleted_provider() {
|
||||
let temp = tempfile::tempdir().expect("tempdir");
|
||||
let path = temp.path().join("models.json");
|
||||
fs::write(
|
||||
&path,
|
||||
r#"{"providers":{"managed":{"api":"before"},"external":{"api":"keep"}}}"#,
|
||||
)
|
||||
.expect("seed");
|
||||
let before =
|
||||
snapshot_pi_provider_values(&path, ["managed".to_string()]).expect("preflight");
|
||||
crate::pi_config::shared_file::fail_next_parent_sync_for_test(&path);
|
||||
|
||||
apply_pi_provider_patch_with_receipt(
|
||||
&path,
|
||||
&before,
|
||||
&IndexMap::from([("managed".to_string(), None)]),
|
||||
)
|
||||
.expect_err("durability failure must remain visible");
|
||||
let restored: Value =
|
||||
serde_json::from_slice(&fs::read(&path).expect("restored document")).expect("parse");
|
||||
assert_eq!(restored["providers"]["managed"]["api"], "before");
|
||||
assert_eq!(restored["providers"]["external"]["api"], "keep");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn external_rename_during_patch_is_reparsed_before_owned_fields_change() {
|
||||
let temp = tempfile::tempdir().expect("tempdir");
|
||||
|
||||
@@ -26,6 +26,9 @@ static BEFORE_COMPARE_EXCHANGE: LazyLock<Mutex<CompareExchangeHooks>> =
|
||||
#[cfg(test)]
|
||||
static BEFORE_ROLLBACK_EXCHANGE: LazyLock<Mutex<CompareExchangeHooks>> =
|
||||
LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||
#[cfg(all(test, unix))]
|
||||
static FAIL_NEXT_PARENT_SYNC: LazyLock<Mutex<HashMap<PathBuf, usize>>> =
|
||||
LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub(crate) struct SharedFileSnapshot {
|
||||
@@ -132,6 +135,20 @@ pub(crate) fn compare_exchange_shared_file_bytes(
|
||||
)
|
||||
}
|
||||
|
||||
/// Retry durability for a namespace state which was observed after a
|
||||
/// compare/exchange returned an error. The caller is still responsible for
|
||||
/// verifying the exact live value before treating the mutation as committed
|
||||
/// or compensated.
|
||||
pub(crate) fn sync_shared_file_parent(path: &Path) -> Result<(), AppError> {
|
||||
let parent = path.parent().ok_or_else(|| {
|
||||
AppError::InvalidInput(format!(
|
||||
"Pi shared-file path has no parent: {}",
|
||||
path.display()
|
||||
))
|
||||
})?;
|
||||
sync_parent(parent)
|
||||
}
|
||||
|
||||
fn compare_exchange_under_lock(
|
||||
path: &Path,
|
||||
expected: Option<&[u8]>,
|
||||
@@ -405,6 +422,13 @@ fn rename_error(
|
||||
|
||||
#[cfg(unix)]
|
||||
fn sync_parent(parent: &Path) -> Result<(), AppError> {
|
||||
#[cfg(test)]
|
||||
if fail_parent_sync_for_test(parent)? {
|
||||
return Err(AppError::io(
|
||||
parent,
|
||||
std::io::Error::other("injected Pi parent-directory sync failure"),
|
||||
));
|
||||
}
|
||||
File::open(parent)
|
||||
.and_then(|directory| directory.sync_all())
|
||||
.map_err(|error| AppError::io(parent, error))
|
||||
@@ -611,6 +635,36 @@ pub(crate) fn replace_before_next_compare_exchange(path: &Path, bytes: &[u8]) {
|
||||
.insert(path.to_path_buf(), bytes.to_vec());
|
||||
}
|
||||
|
||||
#[cfg(all(test, unix))]
|
||||
pub(crate) fn fail_next_parent_sync_for_test(path: &Path) {
|
||||
let parent = path
|
||||
.parent()
|
||||
.expect("a shared-file test path must have a parent")
|
||||
.to_path_buf();
|
||||
let mut failures = FAIL_NEXT_PARENT_SYNC
|
||||
.lock()
|
||||
.expect("parent sync test hook lock");
|
||||
failures
|
||||
.entry(parent)
|
||||
.and_modify(|remaining| *remaining = remaining.saturating_add(1))
|
||||
.or_insert(1);
|
||||
}
|
||||
|
||||
#[cfg(all(test, unix))]
|
||||
fn fail_parent_sync_for_test(parent: &Path) -> Result<bool, AppError> {
|
||||
let mut failures = FAIL_NEXT_PARENT_SYNC
|
||||
.lock()
|
||||
.map_err(|error| AppError::Config(format!("Pi sync test hook is poisoned: {error}")))?;
|
||||
let Some(remaining) = failures.get_mut(parent) else {
|
||||
return Ok(false);
|
||||
};
|
||||
*remaining = remaining.saturating_sub(1);
|
||||
if *remaining == 0 {
|
||||
failures.remove(parent);
|
||||
}
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn run_file_replacement_hook(
|
||||
hooks: &Mutex<CompareExchangeHooks>,
|
||||
|
||||
@@ -58,6 +58,12 @@ pub(crate) struct PiRuntimeSnapshot {
|
||||
pub(crate) server_generation: u64,
|
||||
pub(crate) catalog_epoch: u64,
|
||||
gateway_token: GatewayToken,
|
||||
/// Exact native values which made this immutable runtime reachable.
|
||||
///
|
||||
/// A fenced runtime may only be re-published while these values still
|
||||
/// match `models.json`; retaining the projection alongside the routes
|
||||
/// avoids reconstructing ownership from mutable database/settings state.
|
||||
native_projection: IndexMap<String, Option<Value>>,
|
||||
app_config: AppProxyConfig,
|
||||
providers: HashMap<String, PiRuntimeProvider>,
|
||||
failover_ids: Vec<String>,
|
||||
@@ -366,6 +372,7 @@ pub(crate) fn build_pi_runtime(
|
||||
server_generation,
|
||||
catalog_epoch,
|
||||
gateway_token,
|
||||
native_projection: projection_patch.clone(),
|
||||
app_config,
|
||||
providers,
|
||||
failover_ids,
|
||||
@@ -715,6 +722,18 @@ impl PiRuntimeStore {
|
||||
.map(|snapshot| snapshot.gateway_token.clone())
|
||||
}
|
||||
|
||||
/// Return the exact native projection paired with the fenced runtime.
|
||||
/// This remains available while admission is at an odd epoch so recovery
|
||||
/// can prove that re-publication would still describe Pi's live file.
|
||||
pub(crate) fn retained_native_projection(&self) -> Option<IndexMap<String, Option<Value>>> {
|
||||
self.publication
|
||||
.read()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.current
|
||||
.as_ref()
|
||||
.map(|snapshot| snapshot.native_projection.clone())
|
||||
}
|
||||
|
||||
pub(crate) async fn admission_guard(
|
||||
self: &Arc<Self>,
|
||||
server_generation: u64,
|
||||
|
||||
@@ -1256,6 +1256,27 @@ mod tests {
|
||||
use serde_json::json;
|
||||
use std::sync::Arc;
|
||||
|
||||
struct TestHome(Option<std::ffi::OsString>);
|
||||
|
||||
impl TestHome {
|
||||
fn install(path: &std::path::Path) -> Result<Self, AppError> {
|
||||
let previous = std::env::var_os("CC_SWITCH_TEST_HOME");
|
||||
std::env::set_var("CC_SWITCH_TEST_HOME", path);
|
||||
crate::settings::reload_settings()?;
|
||||
Ok(Self(previous))
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for TestHome {
|
||||
fn drop(&mut self) {
|
||||
match self.0.take() {
|
||||
Some(value) => std::env::set_var("CC_SWITCH_TEST_HOME", value),
|
||||
None => std::env::remove_var("CC_SWITCH_TEST_HOME"),
|
||||
}
|
||||
let _ = crate::settings::reload_settings();
|
||||
}
|
||||
}
|
||||
|
||||
fn managed_input(id: &str, base_url: &str) -> ProviderMutationInput {
|
||||
ProviderMutationInput {
|
||||
id: id.to_string(),
|
||||
@@ -1303,6 +1324,107 @@ mod tests {
|
||||
Ok(config)
|
||||
}
|
||||
|
||||
fn configure_pi_directory(path: &std::path::Path) -> Result<(), AppError> {
|
||||
let mut app_settings = crate::settings::get_settings();
|
||||
app_settings.pi_config_dir = Some(path.to_string_lossy().into_owned());
|
||||
app_settings.pi_takeover_enabled = false;
|
||||
crate::settings::update_settings(app_settings)
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial]
|
||||
#[cfg(unix)]
|
||||
fn create_durability_failure_compensates_database_and_native_projection() -> Result<(), AppError>
|
||||
{
|
||||
let temp = tempfile::tempdir().expect("tempdir");
|
||||
let _home = TestHome::install(temp.path())?;
|
||||
let pi_dir = temp.path().join("pi-agent");
|
||||
configure_pi_directory(&pi_dir)?;
|
||||
let state = AppState::new(Arc::new(Database::memory()?));
|
||||
let models_path = pi_dir.join("models.json");
|
||||
crate::pi_config::shared_file::fail_next_parent_sync_for_test(&models_path);
|
||||
|
||||
let error = PiCatalogCoordinator::apply(
|
||||
&state,
|
||||
PiCatalogMutation::CreateProvider {
|
||||
input: managed_input("ambiguous-create", "https://create.example/v1"),
|
||||
provider_key: "ambiguous-native".to_string(),
|
||||
activate_if_first: false,
|
||||
},
|
||||
)
|
||||
.expect_err("directory durability failure must fail the service operation");
|
||||
assert!(error.to_string().contains("injected"));
|
||||
assert!(state
|
||||
.db
|
||||
.get_provider_aggregate(PI_APP, "ambiguous-create")?
|
||||
.is_none());
|
||||
assert!(state.db.get_pi_projection("ambiguous-create")?.is_none());
|
||||
assert!(
|
||||
!models_path.exists(),
|
||||
"the failed create must not leave a provider or empty shadow file"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial]
|
||||
fn native_import_revalidates_exact_values_before_claiming_success() -> Result<(), AppError> {
|
||||
let temp = tempfile::tempdir().expect("tempdir");
|
||||
let _home = TestHome::install(temp.path())?;
|
||||
let pi_dir = temp.path().join("pi-agent");
|
||||
std::fs::create_dir_all(&pi_dir).expect("Pi directory");
|
||||
configure_pi_directory(&pi_dir)?;
|
||||
let state = AppState::new(Arc::new(Database::memory()?));
|
||||
let models_path = pi_dir.join("models.json");
|
||||
let original = json!({
|
||||
"name": "Native import",
|
||||
"api": "openai-responses",
|
||||
"baseUrl": "https://original.example/v1",
|
||||
"apiKey": "literal-key",
|
||||
"models": [{"id": "model-a", "name": "Model A"}]
|
||||
});
|
||||
std::fs::write(
|
||||
&models_path,
|
||||
serde_json::to_vec_pretty(&json!({"providers": {"native": original}}))
|
||||
.expect("serialize original"),
|
||||
)
|
||||
.expect("write original");
|
||||
let fingerprint = inspect_pi_native_entry(&models_path, "native", &BTreeMap::new())?
|
||||
.expect("native entry")
|
||||
.diagnostic
|
||||
.fingerprint;
|
||||
let external = serde_json::to_vec_pretty(&json!({
|
||||
"providers": {
|
||||
"native": {
|
||||
"name": "External replacement",
|
||||
"api": "openai-responses",
|
||||
"baseUrl": "https://external.example/v1",
|
||||
"apiKey": "external-key",
|
||||
"models": [{"id": "external-model"}]
|
||||
}
|
||||
}
|
||||
}))
|
||||
.expect("serialize external");
|
||||
crate::pi_config::document::replace_before_next_pi_provider_verify(&models_path, &external);
|
||||
|
||||
PiCatalogCoordinator::apply(
|
||||
&state,
|
||||
PiCatalogMutation::ImportNative {
|
||||
provider_key: "native".to_string(),
|
||||
expected_fingerprint: fingerprint,
|
||||
},
|
||||
)
|
||||
.expect_err("an external edit before the final barrier must reject import");
|
||||
assert!(state.db.get_provider_aggregate(PI_APP, "native")?.is_none());
|
||||
assert!(state.db.get_pi_projection("native")?.is_none());
|
||||
assert_eq!(
|
||||
std::fs::read(&models_path).expect("external native file"),
|
||||
external,
|
||||
"the external writer remains authoritative"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn portable_sql_import_claims_only_an_absent_identity_key_and_is_idempotent(
|
||||
) -> Result<(), AppError> {
|
||||
|
||||
@@ -629,6 +629,13 @@ impl ProxyService {
|
||||
}
|
||||
if !crate::settings::pi_takeover_enabled() {
|
||||
self.pi_runtime.close(catalog_epoch).await?;
|
||||
if let Some(expected_native) = expected_native {
|
||||
let models_path = crate::pi_config::native::get_pi_models_path()?;
|
||||
crate::pi_config::document::verify_pi_provider_values(
|
||||
&models_path,
|
||||
&expected_native.values,
|
||||
)?;
|
||||
}
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
let listener = self
|
||||
@@ -702,6 +709,36 @@ impl ProxyService {
|
||||
Ok(build.direct_only_provider_ids)
|
||||
}
|
||||
|
||||
/// Re-open a fenced runtime only while Pi's live exact-key projection
|
||||
/// still matches the immutable snapshot which produced that runtime.
|
||||
///
|
||||
/// Rebuilding this witness from the mutable database or settings would
|
||||
/// turn a failed compensation into stale admission, so the projection is
|
||||
/// retained inside `PiRuntimeSnapshot`.
|
||||
async fn republish_current_pi_runtime_if_native_matches(
|
||||
&self,
|
||||
catalog_epoch: u64,
|
||||
) -> Result<bool, AppError> {
|
||||
let Some(expected_projection) = self.pi_runtime.retained_native_projection() else {
|
||||
self.pi_runtime.close(catalog_epoch).await?;
|
||||
return Ok(false);
|
||||
};
|
||||
let verification = crate::pi_config::native::get_pi_models_path().and_then(|models_path| {
|
||||
crate::pi_config::document::verify_pi_provider_values(
|
||||
&models_path,
|
||||
&expected_projection,
|
||||
)
|
||||
});
|
||||
if let Err(error) = verification {
|
||||
self.pi_runtime.close(catalog_epoch).await?;
|
||||
return Err(AppError::Conflict(format!(
|
||||
"Pi native projection changed while runtime admission was fenced; admission \
|
||||
remains closed: {error}"
|
||||
)));
|
||||
}
|
||||
self.pi_runtime.republish_current(catalog_epoch).await
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn fail_next_pi_reconcile_for_test(&self) {
|
||||
self.fail_next_pi_reconcile.store(true, Ordering::Release);
|
||||
@@ -973,7 +1010,9 @@ impl ProxyService {
|
||||
let old_direct_receipt = match self.restore_pi_direct_projection_at(&old_models_path) {
|
||||
Ok(receipt) => receipt,
|
||||
Err(error) => {
|
||||
let _ = self.pi_runtime.republish_current(epoch).await;
|
||||
let _ = self
|
||||
.republish_current_pi_runtime_if_native_matches(epoch)
|
||||
.await;
|
||||
return Err(error);
|
||||
}
|
||||
};
|
||||
@@ -1019,7 +1058,9 @@ impl ProxyService {
|
||||
.await
|
||||
.is_ok()
|
||||
} else {
|
||||
self.pi_runtime.republish_current(epoch).await.is_ok()
|
||||
self.republish_current_pi_runtime_if_native_matches(epoch)
|
||||
.await
|
||||
.unwrap_or(false)
|
||||
};
|
||||
return Err(AppError::Config(format!(
|
||||
"failed to reconcile Pi prompts in the new directory: {error}; rollback: settings={settings_restored}, prompts={prompts_restored}, skills={skills_restored}, gateway={runtime_restored}"
|
||||
@@ -1048,7 +1089,9 @@ impl ProxyService {
|
||||
.await
|
||||
.is_ok()
|
||||
} else {
|
||||
self.pi_runtime.republish_current(epoch).await.is_ok()
|
||||
self.republish_current_pi_runtime_if_native_matches(epoch)
|
||||
.await
|
||||
.unwrap_or(false)
|
||||
};
|
||||
return Err(AppError::Config(format!(
|
||||
"failed to reconcile Pi Skills in the new directory: {error}; rollback: settings={settings_restored}, prompts={prompts_restored}, skills={skills_restored}, gateway={runtime_restored}"
|
||||
@@ -1078,7 +1121,9 @@ impl ProxyService {
|
||||
.await
|
||||
.is_ok()
|
||||
} else {
|
||||
self.pi_runtime.republish_current(epoch).await.is_ok()
|
||||
self.republish_current_pi_runtime_if_native_matches(epoch)
|
||||
.await
|
||||
.unwrap_or(false)
|
||||
};
|
||||
return Err(AppError::Config(format!(
|
||||
"failed to publish Pi in the new native directory: {error}; rollback: settings={settings_restored}, prompts={prompts_restored}, skills={skills_restored}, old_gateway={old_gateway_restored}"
|
||||
@@ -1096,7 +1141,9 @@ impl ProxyService {
|
||||
let direct_receipt = match self.restore_pi_direct_projection() {
|
||||
Ok(receipt) => receipt,
|
||||
Err(error) => {
|
||||
let _ = self.pi_runtime.republish_current(epoch).await;
|
||||
let _ = self
|
||||
.republish_current_pi_runtime_if_native_matches(epoch)
|
||||
.await;
|
||||
return Err(error);
|
||||
}
|
||||
};
|
||||
@@ -1761,8 +1808,10 @@ impl ProxyService {
|
||||
if let Err(error) = self.reconcile_pi_runtime().await {
|
||||
if was_enabled {
|
||||
let epoch = self.pi_runtime.begin_mutation().await;
|
||||
let restored = self.pi_runtime.republish_current(epoch).await;
|
||||
return Err(if restored.is_ok() {
|
||||
let restored = self
|
||||
.republish_current_pi_runtime_if_native_matches(epoch)
|
||||
.await;
|
||||
return Err(if matches!(restored, Ok(true)) {
|
||||
format!(
|
||||
"failed to refresh Pi gateway catalog; the previous runtime remains active: {error}"
|
||||
)
|
||||
@@ -1814,7 +1863,9 @@ impl ProxyService {
|
||||
let direct_receipt = match self.restore_pi_direct_projection() {
|
||||
Ok(receipt) => receipt,
|
||||
Err(error) => {
|
||||
let _ = self.pi_runtime.republish_current(epoch).await;
|
||||
let _ = self
|
||||
.republish_current_pi_runtime_if_native_matches(epoch)
|
||||
.await;
|
||||
return Err(format!(
|
||||
"failed to restore Pi's direct native projection: {error}"
|
||||
));
|
||||
@@ -5234,6 +5285,112 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn fenced_pi_runtime_is_not_republished_after_external_native_drift() {
|
||||
let home = TempHome::new();
|
||||
crate::settings::reload_settings().expect("reload isolated settings");
|
||||
let pi_dir = home.dir.path().join("pi");
|
||||
std::fs::create_dir_all(&pi_dir).expect("Pi directory");
|
||||
let mut settings = crate::settings::get_settings();
|
||||
settings.pi_config_dir = Some(pi_dir.to_string_lossy().into_owned());
|
||||
settings.pi_takeover_enabled = true;
|
||||
crate::settings::update_settings(settings).expect("Pi settings");
|
||||
|
||||
let db = Arc::new(Database::memory().expect("database"));
|
||||
let direct = json!({
|
||||
"name": "Managed Pi",
|
||||
"api": "openai-responses",
|
||||
"baseUrl": "https://managed.example/v1",
|
||||
"apiKey": "managed-key",
|
||||
"models": [{"id": "model-a", "name": "Model A"}]
|
||||
});
|
||||
db.create_pi_catalog_provider(
|
||||
NewProviderAggregate::from_input(
|
||||
AppType::Pi.as_str(),
|
||||
ProviderMutationInput {
|
||||
id: "managed-pi".to_string(),
|
||||
name: "Managed Pi".to_string(),
|
||||
settings_config: direct.clone(),
|
||||
website_url: None,
|
||||
category: None,
|
||||
created_at: None,
|
||||
sort_index: Some(0),
|
||||
notes: None,
|
||||
meta: None,
|
||||
icon: Some("pi".to_string()),
|
||||
icon_color: None,
|
||||
in_failover_queue: false,
|
||||
},
|
||||
)
|
||||
.expect("aggregate"),
|
||||
"managed-pi",
|
||||
)
|
||||
.expect("managed provider");
|
||||
let models_path = pi_dir.join("models.json");
|
||||
std::fs::write(
|
||||
&models_path,
|
||||
serde_json::to_vec_pretty(&json!({"providers": {"managed-pi": direct}}))
|
||||
.expect("direct models"),
|
||||
)
|
||||
.expect("seed direct models");
|
||||
|
||||
let service = ProxyService::new(db);
|
||||
*service
|
||||
.pi_listener
|
||||
.write()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some(PiListenerIdentity {
|
||||
server_generation: 41,
|
||||
gateway_origin: url::Url::parse("http://127.0.0.1:15721/").expect("gateway origin"),
|
||||
});
|
||||
service
|
||||
.reconcile_pi_runtime()
|
||||
.await
|
||||
.expect("publish initial runtime");
|
||||
assert_eq!(
|
||||
service
|
||||
.get_takeover_status()
|
||||
.await
|
||||
.expect("active status")
|
||||
.pi_operational_state,
|
||||
PiTakeoverOperationalState::Active
|
||||
);
|
||||
|
||||
let epoch = service.begin_pi_catalog_mutation().await;
|
||||
let external = serde_json::to_vec_pretty(&json!({
|
||||
"providers": {
|
||||
"managed-pi": {
|
||||
"name": "External",
|
||||
"api": "openai-responses",
|
||||
"baseUrl": "https://external.example/v1",
|
||||
"apiKey": "external-key",
|
||||
"models": [{"id": "external-model"}]
|
||||
}
|
||||
}
|
||||
}))
|
||||
.expect("external models");
|
||||
std::fs::write(&models_path, &external).expect("external edit");
|
||||
|
||||
assert!(matches!(
|
||||
service
|
||||
.republish_current_pi_runtime_if_native_matches(epoch)
|
||||
.await,
|
||||
Err(AppError::Conflict(_))
|
||||
));
|
||||
assert_eq!(
|
||||
service
|
||||
.get_takeover_status()
|
||||
.await
|
||||
.expect("degraded status")
|
||||
.pi_operational_state,
|
||||
PiTakeoverOperationalState::Degraded
|
||||
);
|
||||
assert_eq!(
|
||||
std::fs::read(&models_path).expect("external file remains"),
|
||||
external
|
||||
);
|
||||
}
|
||||
|
||||
fn seed_codex_model_template() {
|
||||
let codex_dir = crate::codex_config::get_codex_config_dir();
|
||||
std::fs::create_dir_all(&codex_dir).expect("create codex dir");
|
||||
|
||||
Reference in New Issue
Block a user