From cccd6354903637b66f9f77102b8186d7286ca655 Mon Sep 17 00:00:00 2001 From: Microck Date: Sat, 5 Sep 2026 10:04:07 +0000 Subject: [PATCH] fix: truncate locked cache file after acquiring the lock --- Cargo.lock | 33 ++++++ Cargo.toml | 1 + src/cli.rs | 43 ++++++- src/local.rs | 237 +++++++++++++++++++++++++++++++++++---- src/main.rs | 93 ++++++++++----- tests/integration-cli.rs | 156 ++++++++++++++++++++++++++ 6 files changed, 511 insertions(+), 52 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 94d0160..d9b9d31 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -561,6 +561,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "futures-channel" version = "0.3.32" @@ -1062,6 +1072,7 @@ dependencies = [ "cliclack", "console", "ctrlc", + "fs2", "futures-util", "httpmock", "jsonc-parser", @@ -2483,6 +2494,28 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + [[package]] name = "windows-link" version = "0.2.1" diff --git a/Cargo.toml b/Cargo.toml index d738462..96160a2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,6 +30,7 @@ console = "0.16.3" ctrlc = "3.5.2" reqwest = { version = "0.12.15", default-features = false, features = ["brotli", "deflate", "gzip", "json", "multipart", "rustls-tls", "stream"] } futures-util = "0.3.34" +fs2 = "0.4" jsonc-parser = { version = "0.33.1", features = ["serde"] } scraper = "0.27.0" serde = { version = "1.0.228", features = ["derive"] } diff --git a/src/cli.rs b/src/cli.rs index a9134bf..2aab176 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -870,6 +870,20 @@ pub struct FastGptArgs { #[arg(long, value_name = "SECONDS")] pub cache_ttl: Option, } +impl FastGptArgs { + /// Validates fastgpt arguments. + /// + /// # Errors + /// Returns an error when web search is explicitly disabled, which the + /// upstream FastGPT API does not support. + pub fn validate(&self) -> Result<(), String> { + if self.web_search == Some(false) { + return Err("fastgpt --web-search false is not supported by the upstream FastGPT API, which requires web search grounding. Omit --web-search or pass --web-search true".to_string()); + } + + Ok(()) + } +} #[derive(Debug, Args)] /// Arguments for the `news` subcommand. @@ -1920,8 +1934,8 @@ pub struct RedirectUpdateArgs { #[cfg(test)] mod tests { use super::{ - Cli, Commands, NewsArgs, NewsFilterMode, NewsFilterScope, OutputFormat, SearchArgs, - SearchOrder, SearchTime, SummarizeArgs, + Cli, Commands, FastGptArgs, NewsArgs, NewsFilterMode, NewsFilterScope, OutputFormat, + SearchArgs, SearchOrder, SearchTime, SummarizeArgs, }; use clap::Parser; @@ -2074,6 +2088,31 @@ mod tests { .expect_err("summarize should require url or text input"); assert!(error.contains("exactly one of --url or --text")); } + #[test] + fn rejects_fastgpt_disabled_web_search() { + let args = FastGptArgs { + query: "What is the capital of Australia?".to_string(), + cache: None, + web_search: Some(false), + local_cache: false, + cache_ttl: None, + }; + let error = args + .validate() + .expect_err("web_search false should be rejected"); + assert!(error.contains("not supported")); + + for web_search in [None, Some(true)] { + let args = FastGptArgs { + query: "What is the capital of Australia?".to_string(), + cache: None, + web_search, + local_cache: false, + cache_ttl: None, + }; + assert!(args.validate().is_ok()); + } + } #[test] fn parses_product_workflow_commands() { diff --git a/src/local.rs b/src/local.rs index a1c0dd6..49f5904 100644 --- a/src/local.rs +++ b/src/local.rs @@ -4,15 +4,23 @@ //! directory. The issue asks for local, opt-in workflow state; a file-backed //! store keeps that state inspectable and avoids adding a database dependency //! before the CLI has enough history volume to justify it. +//! +//! Every file access takes an advisory lock (`fs2`) so concurrent CLI +//! processes cannot interleave writes or read torn files. Mutations that +//! read, modify, and rewrite a file hold one exclusive lock across the whole +//! operation; see [`update_site_preferences`]. use std::collections::BTreeMap; use std::collections::hash_map::DefaultHasher; use std::env; use std::fs; +use std::fs::File; use std::hash::{Hash, Hasher}; +use std::io::{Read, Seek, Write}; use std::path::{Path, PathBuf}; use std::time::{SystemTime, UNIX_EPOCH}; +use fs2::FileExt; use serde::{Deserialize, Serialize}; use serde_json::Value; @@ -94,7 +102,9 @@ pub fn cache_get(key: &str) -> Result, KagiError> { return Ok(None); } - let raw = fs::read_to_string(&path).map_err(|error| { + let mut file = open_locked_shared(&path)?; + let mut raw = String::new(); + file.read_to_string(&mut raw).map_err(|error| { KagiError::Config(format!( "failed to read cache entry {}: {error}", path.display() @@ -118,13 +128,12 @@ pub fn cache_get(key: &str) -> Result, KagiError> { pub fn cache_put(key: &str, ttl_seconds: u64, value: &Value) -> Result<(), KagiError> { let path = cache_response_path(key); - ensure_parent_dir(&path)?; let envelope = CacheEnvelope { created_at: now_unix_seconds()?, ttl_seconds, value: value.clone(), }; - write_json(&path, &envelope) + write_json_locked(&path, &envelope) } pub fn append_history(entry: &HistoryEntry) -> Result<(), KagiError> { @@ -132,20 +141,13 @@ pub fn append_history(entry: &HistoryEntry) -> Result<(), KagiError> { ensure_parent_dir(&path)?; let mut raw = serde_json::to_string(entry)?; raw.push('\n'); - fs::OpenOptions::new() - .create(true) - .append(true) - .open(&path) - .and_then(|mut file| { - use std::io::Write; - file.write_all(raw.as_bytes()) - }) - .map_err(|error| { - KagiError::Config(format!( - "failed to append history {}: {error}", - path.display() - )) - }) + let mut file = open_locked_append(&path)?; + file.write_all(raw.as_bytes()).map_err(|error| { + KagiError::Config(format!( + "failed to append history {}: {error}", + path.display() + )) + }) } pub fn read_history(limit: usize) -> Result, KagiError> { @@ -154,7 +156,9 @@ pub fn read_history(limit: usize) -> Result, KagiError> { return Ok(vec![]); } - let raw = fs::read_to_string(&path).map_err(|error| { + let mut file = open_locked_shared(&path)?; + let mut raw = String::new(); + file.read_to_string(&mut raw).map_err(|error| { KagiError::Config(format!( "failed to read history {}: {error}", path.display() @@ -201,7 +205,9 @@ pub fn load_site_preferences() -> Result { if !path.exists() { return Ok(SitePreferences::default()); } - let raw = fs::read_to_string(&path).map_err(|error| { + let mut file = open_locked_shared(&path)?; + let mut raw = String::new(); + file.read_to_string(&mut raw).map_err(|error| { KagiError::Config(format!( "failed to read site preferences {}: {error}", path.display() @@ -215,10 +221,72 @@ pub fn load_site_preferences() -> Result { }) } -pub fn save_site_preferences(preferences: &SitePreferences) -> Result<(), KagiError> { +/// Reads, mutates, and rewrites the site preferences file while holding one +/// exclusive advisory lock, so concurrent `site-pref set` and `site-pref +/// remove` invocations cannot silently lose each other's updates. Prefer +/// this over separate [`load_site_preferences`] reads followed by blind +/// rewrites whenever a mutation depends on the previously stored state. +pub fn update_site_preferences( + update: impl FnOnce(&mut SitePreferences) -> Result, +) -> Result { let path = site_preferences_path(); ensure_parent_dir(&path)?; - write_json(&path, preferences) + let mut file = fs::OpenOptions::new() + .create(true) + .read(true) + .write(true) + .truncate(false) + .open(&path) + .map_err(|error| { + KagiError::Config(format!( + "failed to open site preferences {}: {error}", + path.display() + )) + })?; + file.lock_exclusive().map_err(|error| { + KagiError::Config(format!( + "failed to lock site preferences {}: {error}", + path.display() + )) + })?; + let mut raw = String::new(); + file.read_to_string(&mut raw).map_err(|error| { + KagiError::Config(format!( + "failed to read site preferences {}: {error}", + path.display() + )) + })?; + let mut preferences = if raw.trim().is_empty() { + SitePreferences::default() + } else { + serde_json::from_str(&raw).map_err(|error| { + KagiError::Parse(format!( + "failed to parse site preferences {}: {error}", + path.display() + )) + })? + }; + let result = update(&mut preferences)?; + let raw = serde_json::to_string_pretty(&preferences)?; + file.set_len(0).map_err(|error| { + KagiError::Config(format!( + "failed to write site preferences {}: {error}", + path.display() + )) + })?; + file.rewind().map_err(|error| { + KagiError::Config(format!( + "failed to write site preferences {}: {error}", + path.display() + )) + })?; + file.write_all(raw.as_bytes()).map_err(|error| { + KagiError::Config(format!( + "failed to write site preferences {}: {error}", + path.display() + )) + })?; + Ok(result) } pub fn normalize_domain(input: &str) -> Result { @@ -255,12 +323,59 @@ fn site_preferences_path() -> PathBuf { cache_root().join("site-preferences.json") } -fn write_json(path: &Path, value: &T) -> Result<(), KagiError> { - let raw = serde_json::to_string_pretty(value)?; - fs::write(path, raw) +/// Serializes `value` as pretty JSON and writes it to `path` while holding +/// an exclusive advisory lock, so readers under a shared lock never observe +/// a partially written file. +fn write_json_locked(path: &Path, value: &T) -> Result<(), KagiError> { + ensure_parent_dir(path)?; + let mut file = fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(false) + .open(path) + .map_err(|error| { + KagiError::Config(format!("failed to write {}: {error}", path.display())) + })?; + file.lock_exclusive().map_err(|error| { + KagiError::Config(format!("failed to lock {}: {error}", path.display())) + })?; + // Truncate only after the lock is held: truncating in `open()` would + // empty the file while a concurrent shared-lock reader is still reading. + file.set_len(0).map_err(|error| { + KagiError::Config(format!("failed to write {}: {error}", path.display())) + })?; + serde_json::to_writer_pretty(&file, value)?; + file.flush() .map_err(|error| KagiError::Config(format!("failed to write {}: {error}", path.display()))) } +/// Opens `path` in append mode and acquires an exclusive advisory lock, so +/// one JSON Lines record is always written without interleaving. +fn open_locked_append(path: &Path) -> Result { + let file = fs::OpenOptions::new() + .create(true) + .append(true) + .open(path) + .map_err(|error| { + KagiError::Config(format!("failed to append {}: {error}", path.display())) + })?; + file.lock_exclusive().map_err(|error| { + KagiError::Config(format!("failed to lock {}: {error}", path.display())) + })?; + Ok(file) +} + +/// Opens an existing `path` for reading and acquires a shared advisory lock. +fn open_locked_shared(path: &Path) -> Result { + let file = fs::File::open(path).map_err(|error| { + KagiError::Config(format!("failed to read {}: {error}", path.display())) + })?; + file.lock_shared().map_err(|error| { + KagiError::Config(format!("failed to lock {}: {error}", path.display())) + })?; + Ok(file) +} + fn ensure_parent_dir(path: &Path) -> Result<(), KagiError> { if let Some(parent) = path.parent() { fs::create_dir_all(parent).map_err(|error| { @@ -274,6 +389,7 @@ fn ensure_parent_dir(path: &Path) -> Result<(), KagiError> { mod tests { use super::*; use crate::test_support::lock_env; + use std::sync::{Arc, Barrier}; use tempfile::TempDir; #[test] @@ -297,4 +413,77 @@ mod tests { ); assert!(normalize_domain(" ").is_err()); } + + #[test] + fn concurrent_preference_updates_all_survive() { + let _guard = lock_env(); + let tempdir = TempDir::new().expect("tempdir"); + unsafe { env::set_var(CACHE_DIR_ENV, tempdir.path()) }; + + let domains: Vec = (0..16) + .map(|index| format!("domain-{index}.example")) + .collect(); + let barrier = Arc::new(Barrier::new(domains.len())); + let mut handles = Vec::new(); + for domain in domains.clone() { + let barrier = Arc::clone(&barrier); + handles.push(std::thread::spawn(move || { + barrier.wait(); + update_site_preferences(|preferences| { + preferences + .domains + .insert(domain.clone(), SitePreferenceMode::Pin); + Ok::<(), KagiError>(()) + }) + .expect("concurrent preference update"); + })); + } + for handle in handles { + handle.join().expect("worker thread"); + } + + let stored = load_site_preferences().expect("load preferences"); + for domain in &domains { + assert_eq!( + stored.domains.get(domain), + Some(&SitePreferenceMode::Pin), + "concurrent update for {domain} must survive" + ); + } + unsafe { env::remove_var(CACHE_DIR_ENV) }; + } + + #[test] + fn concurrent_history_appends_stay_parseable() { + let _guard = lock_env(); + let tempdir = TempDir::new().expect("tempdir"); + unsafe { env::set_var(CACHE_DIR_ENV, tempdir.path()) }; + + let workers = 8; + let per_worker = 25; + let barrier = Arc::new(Barrier::new(workers)); + let mut handles = Vec::new(); + for worker in 0..workers { + let barrier = Arc::clone(&barrier); + handles.push(std::thread::spawn(move || { + barrier.wait(); + for index in 0..per_worker { + append_history(&HistoryEntry { + timestamp: (worker * per_worker + index) as u64, + command: "search".to_string(), + query: Some(format!("worker {worker} query {index}")), + result_count: Some(index), + }) + .expect("concurrent history append"); + } + })); + } + for handle in handles { + handle.join().expect("worker thread"); + } + + let entries = read_history(0).expect("history must stay parseable"); + assert_eq!(entries.len(), workers * per_worker); + unsafe { env::remove_var(CACHE_DIR_ENV) }; + } } diff --git a/src/main.rs b/src/main.rs index 6f7dcc4..92b57f3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -281,9 +281,12 @@ async fn run() -> Result<(), KagiError> { let cli = Cli::parse(); if cli.generate_completion.is_some() && cli.command.is_some() { - return Err(KagiError::Config( - "completion was not generated because --generate-completion cannot be used with a command. Run `kagi --generate-completion ` by itself".to_string(), - )); + Cli::command() + .error( + clap::error::ErrorKind::ArgumentConflict, + "completion was not generated because --generate-completion cannot be used with a command. Run `kagi --generate-completion ` by itself", + ) + .exit(); } if let Some(shell) = cli.generate_completion { @@ -724,6 +727,7 @@ async fn run() -> Result<(), KagiError> { print_json(&response) } Commands::Fastgpt(args) => { + args.validate().map_err(KagiError::Config)?; let request = FastGptRequest { query: args.query, cache: args.cache, @@ -2890,24 +2894,60 @@ async fn run_summarize_filter( )); } - let mut results = Vec::new(); + let mut failure_count = 0usize; if args.subscriber { let token = resolve_session_token(profile)?; for item in lines { let request = summarize_item_request_subscriber(&item, &args); - let response = execute_subscriber_summarize(&request, &token).await?; - results.push(serde_json::json!({ "input": item, "response": response })); + match execute_subscriber_summarize(&request, &token).await { + Ok(response) => print_summarize_filter_item(&item, true, response)?, + Err(error) => { + failure_count += 1; + print_summarize_filter_item(&item, false, error_envelope(&error))?; + } + } } } else { let token = resolve_api_token(profile)?; for item in lines { let request = summarize_item_request_public(&item, &args); - let response = execute_summarize(&request, &token).await?; - results.push(serde_json::json!({ "input": item, "response": response })); + match execute_summarize(&request, &token).await { + Ok(response) => print_summarize_filter_item(&item, true, response)?, + Err(error) => { + failure_count += 1; + print_summarize_filter_item(&item, false, error_envelope(&error))?; + } + } } } - print_json(&serde_json::json!({ "results": results })) + if failure_count > 0 { + return Err(KagiError::Batch(format!( + "summarize --filter completed with {failure_count} failed item(s)" + ))); + } + + Ok(()) +} + +fn print_summarize_filter_item( + item: &str, + ok: bool, + payload: impl serde::Serialize, +) -> Result<(), KagiError> { + if ok { + print_compact_json(&serde_json::json!({ + "input": item, + "ok": true, + "response": payload, + })) + } else { + print_compact_json(&serde_json::json!({ + "input": item, + "ok": false, + "error": payload, + })) + } } fn summarize_item_request_subscriber( @@ -3045,21 +3085,20 @@ fn run_site_pref(command: SitePrefSubcommand) -> Result<(), KagiError> { match command { SitePrefSubcommand::List => print_json(&local::load_site_preferences()?), SitePrefSubcommand::Set(args) => { - let mut preferences = local::load_site_preferences()?; let domain = local::normalize_domain(&args.domain)?; - preferences - .domains - .insert(domain.clone(), site_pref_mode(args.mode)); - local::save_site_preferences(&preferences)?; - print_json( - &serde_json::json!({ "domain": domain, "mode": site_pref_mode(args.mode).as_str() }), - ) + let mode = site_pref_mode(args.mode); + local::update_site_preferences(|preferences| { + preferences.domains.insert(domain.clone(), mode); + Ok::<(), KagiError>(()) + })?; + print_json(&serde_json::json!({ "domain": domain, "mode": mode.as_str() })) } SitePrefSubcommand::Remove(args) => { - let mut preferences = local::load_site_preferences()?; let domain = local::normalize_domain(&args.domain)?; - preferences.domains.remove(&domain); - local::save_site_preferences(&preferences)?; + local::update_site_preferences(|preferences| { + preferences.domains.remove(&domain); + Ok::<(), KagiError>(()) + })?; print_json(&serde_json::json!({ "domain": domain, "removed": true })) } } @@ -4685,21 +4724,23 @@ async fn mcp_mutating_tool_call( config.default_output_or(OutputFormat::Json), ), "kagi_site_pref_set" => { - let mut preferences = local::load_site_preferences()?; let domain = local::normalize_domain(&mcp_required_string(arguments, "domain")?)?; let mode = mcp_site_pref_mode(arguments)?; - preferences.domains.insert(domain.clone(), mode); - local::save_site_preferences(&preferences)?; + local::update_site_preferences(|preferences| { + preferences.domains.insert(domain.clone(), mode); + Ok::<(), KagiError>(()) + })?; mcp_output( &serde_json::json!({"domain": domain, "mode": mode.as_str()}), config.default_output_or(OutputFormat::Json), ) } "kagi_site_pref_remove" => { - let mut preferences = local::load_site_preferences()?; let domain = local::normalize_domain(&mcp_required_string(arguments, "domain")?)?; - preferences.domains.remove(&domain); - local::save_site_preferences(&preferences)?; + local::update_site_preferences(|preferences| { + preferences.domains.remove(&domain); + Ok::<(), KagiError>(()) + })?; mcp_output( &serde_json::json!({"domain": domain, "removed": true}), config.default_output_or(OutputFormat::Json), diff --git a/tests/integration-cli.rs b/tests/integration-cli.rs index f8f42c9..32e2fcf 100644 --- a/tests/integration-cli.rs +++ b/tests/integration-cli.rs @@ -3931,3 +3931,159 @@ fn mcp_tool_call_error_returns_json_rpc_error_and_keeps_server_alive() { "expected successful result for second tool call, got: {success_resp}" ); } +#[test] +fn generate_completion_with_subcommand_exits_usage_error() { + let tempdir = TempDir::new().expect("tempdir"); + let output = run_kagi( + &["--generate-completion", "bash", "auth", "status"], + &[], + tempdir.path(), + ); + + assert_eq!( + output.status.code(), + Some(2), + "usage conflict should exit 2, got {:?}\nstdout:\n{}\nstderr:\n{}", + output.status.code(), + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + String::from_utf8_lossy(&output.stderr).contains("cannot be used with a command"), + "expected conflict explanation on stderr, got:\n{}", + String::from_utf8_lossy(&output.stderr) + ); +} + +#[test] +fn fastgpt_rejects_disabled_web_search_before_network() { + let tempdir = TempDir::new().expect("tempdir"); + let output = run_kagi( + &[ + "fastgpt", + "--web-search", + "false", + "What is the capital of Australia?", + ], + &[], + tempdir.path(), + ); + + assert!( + !output.status.success(), + "web_search false should fail, got {:?}", + output.status.code() + ); + assert!( + String::from_utf8_lossy(&output.stderr).contains("not supported"), + "expected upstream support explanation on stderr, got:\n{}", + String::from_utf8_lossy(&output.stderr) + ); +} + +#[test] +fn summarize_filter_streams_per_item_envelopes() { + let server = MockServer::start(); + let _good = server.mock(|when, then| { + when.method(POST) + .path("/api/v0/summarize") + .header("authorization", "Bot test-api-token") + .body_includes("good.example"); + then.status(200) + .header("content-type", "application/json") + .json_body(json!({ + "meta": api_meta(), + "data": { + "output": "A concise summary.", + "tokens": 42 + } + })); + }); + let _bad = server.mock(|when, then| { + when.method(POST) + .path("/api/v0/summarize") + .header("authorization", "Bot test-api-token") + .body_includes("bad.example"); + then.status(500) + .header("content-type", "application/json") + .json_body(json!({ + "meta": api_meta(), + "error": { "code": "internal_error", "msg": "boom" } + })); + }); + + let tempdir = TempDir::new().expect("tempdir"); + let env = test_env(&server); + let output = run_kagi_with_stdin( + &["summarize", "--filter"], + "https://good.example/article\nhttps://bad.example/article\n", + &env_refs(&env), + tempdir.path(), + ); + + assert_eq!( + output.status.code(), + Some(1), + "expected exit 1 when one filter item fails, got {:?}", + output.status.code() + ); + let records = String::from_utf8_lossy(&output.stdout) + .lines() + .map(|line| serde_json::from_str::(line).expect("JSONL record should parse")) + .collect::>(); + assert_eq!(records.len(), 2, "both items must be processed"); + assert_eq!(records[0]["input"], "https://good.example/article"); + assert_eq!(records[0]["ok"], true); + assert_eq!( + records[0]["response"]["data"]["output"], + "A concise summary." + ); + assert_eq!(records[1]["input"], "https://bad.example/article"); + assert_eq!(records[1]["ok"], false); + assert!( + records[1]["error"].is_object(), + "failed item must carry an error envelope, got: {}", + records[1] + ); + assert!( + String::from_utf8_lossy(&output.stderr) + .contains("summarize --filter completed with 1 failed item"), + "expected aggregate failure on stderr, got:\n{}", + String::from_utf8_lossy(&output.stderr) + ); +} + +#[test] +fn concurrent_site_pref_sets_preserve_all_domains() { + let tempdir = TempDir::new().expect("tempdir"); + let dir = tempdir.path().to_path_buf(); + let mut handles = Vec::new(); + for index in 0..8 { + let dir = dir.clone(); + handles.push(std::thread::spawn(move || { + let domain = format!("domain-{index}.example"); + run_kagi(&["site-pref", "set", &domain, "--mode", "pin"], &[], &dir) + })); + } + for handle in handles { + let output = handle.join().expect("worker thread"); + assert!( + output.status.success(), + "concurrent site-pref set should succeed, got {:?}\nstderr:\n{}", + output.status.code(), + String::from_utf8_lossy(&output.stderr) + ); + } + + let list = run_kagi(&["site-pref", "list"], &[], &dir); + assert_success(&list); + let prefs: Value = serde_json::from_slice(&list.stdout).expect("prefs json parses"); + for index in 0..8 { + let domain = format!("domain-{index}.example"); + assert_eq!( + prefs["domains"][domain.as_str()], + "pin", + "concurrent update for {domain} must survive" + ); + } +}