diff --git a/OPTIMIZATION_TODOS.md b/OPTIMIZATION_TODOS.md index 1bc8cf5..225e59a 100644 --- a/OPTIMIZATION_TODOS.md +++ b/OPTIMIZATION_TODOS.md @@ -1,78 +1,104 @@ # Optimization TODO list -Full codebase review after the SQLite event-bus migration (`7915613`). -Priorities: **P0** = correctness / unbounded growth; **P1** = hot-path performance; -**P2** = medium impact; **P3** = polish / future. +Full codebase review after project repository queries / RPC routing +(`aaacba9`), plus unmerged hot-path fixes from earlier review rounds. +Priorities: **P0** = correctness / unbounded growth; **P1** = hot-path +performance; **P2** = medium impact; **P3** = polish / future. Stub crates (`retcon-worktrees`, `retcon-browser`, `retcon-checkpoints`, `retcon-filesystem`, `retcon-index`, `retcon-platform`, `retcon-permissions`, -`retcon-diagnostics`, `retcon-updater`) and empty TypeScript packages have no -meaningful optimization surface yet. +`retcon-diagnostics`, `retcon-updater`) and empty TypeScript/Dart packages +have no meaningful optimization surface yet. --- ## Done in this PR -- [x] **P0** Cap browser-service console/network buffers with a ring buffer - (`apps/browser-service/src/browser.ts`) -- [x] **P0** Remove timed-out / closed browser RPC entries from `pending` - (`crates/retcon-core/src/spikes/browser.rs`) -- [x] **P0** Cap RPC frame size on read (`crates/retcon-core/src/frame.rs`, - `server.rs`, `spikes/browser.rs`) -- [x] **P0** Periodic SQLite event retention (`crates/retcon-core/src/event.rs`) -- [x] **P0** Evict finished jobs from memory + DB (`crates/retcon-core/src/jobs.rs`) -- [x] **P0** Graceful browser-service shutdown before kill (`spikes/browser.rs`) -- [x] **P0** Bound Git command output (`crates/retcon-git/src/lib.rs`) -- [x] **P1** Don't hold event-bus mutex across SQLite insert (`event.rs` writer thread) -- [x] **P1** Batch / coalesce terminal output events (`spikes/terminal.rs`) -- [x] **P1** Move SQLite work off Tokio worker threads (`database.rs` `read_async`, - `artifacts.rs` async helpers) -- [x] **P1** Add status indexes for recovery queries (migration `0003`) -- [x] **P1** Enforce log pagination / tail in `browser.logs` (`browser.ts`) -- [x] **P1** Default screenshot to viewport (`browser.ts`) -- [x] **P1** Combine Git status into one process (`retcon-git/src/lib.rs`) -- [x] **P2** Browser proc lock held across stdin writes — writer actor split (`browser.rs`) -- [x] **P2** Clear stale `BrowserHandle.proc` + drain pending on reader exit (`browser.rs`) -- [x] **P2** Artifact retention hash indexes (migration `0003`) -- [x] **P2** Run artifact disk walks in `spawn_blocking` (`artifacts.rs`) -- [x] **P2** Skip full rehash on duplicate artifact store (`artifacts.rs`) -- [x] **P2** Binary-search event replay (`event.rs`) -- [x] **P2** `Arc` to cut clone cost on emit (`event.rs`) -- [x] **P2** `prepare_cached` for hot statements (`event.rs`, `jobs.rs`) -- [x] **P2** `PRAGMA quick_check` on normal open; full check for diagnostics (`database.rs`) -- [x] **P2** `SQLITE_OPEN_NO_MUTEX` when Rust serializes the connection (`database.rs`) -- [x] **P2** Tune `cache_size` / `mmap_size` / `wal_autocheckpoint` (`database.rs`) -- [x] **P2** Debounce job persistence; avoid Debug-format enum names (`jobs.rs`) -- [x] **P2** Don't hold terminal registry lock while killing (`spikes/terminal.rs`) -- [x] **P2** Cache / async shell detection (`spikes/terminal.rs`) -- [x] **P2** Cache agent/provider detection (`retcon-agents`) -- [x] **P2** Await `child.wait()` instead of 250ms poll (`spikes/agent.rs`) -- [x] **P2** Cache schema version after open (`state.rs`) -- [x] **P2** Lazy-load event replay pages from DB at startup (`event.rs`) -- [x] **P2** Pass browser temp profile dir via `launchPersistentContext` (`browser.ts`) -- [x] **P2** Stdout backpressure on browser-service event emits (`main.ts`) -- [x] **P2** Spawn `bun` directly (no `cmd /C` shell) on Windows (`spikes/browser.rs`) -- [x] **P3** Deduplicate console/network stream vs buffer shipping (`browser.ts` — network - responses now land in the ring buffer) -- [x] **P3** Method-aware concurrency for browser-service stdio RPC (`main.ts`) -- [x] **P3** Cache log level at module load (`logging.ts`) -- [x] **P3** Website asset generation: single Sharp pipeline + `Promise.all` for icons -- [x] **P3** Website CLS: explicit image dimensions; defer Tally iframe (`index.astro`) -- [x] **P3** CI-generate PWA icons referenced by `site.webmanifest` (`package.json` build) +### From project RPC commit (`aaacba9`) + +- [x] **P0** Cap `project.clone` / project Git helpers via `retcon_git` + (no unbounded `.output()`) — `crates/retcon-core/src/projects.rs` +- [x] **P1** Avoid full `git status --porcelain` in health; use + `rev-parse` + `diff --quiet` + capped `ls-files` — + `projects.rs` +- [x] **P1** Run filesystem `analyze()` in `spawn_blocking` — + `projects.rs` +- [x] **P1** Bound `project.list` (default 50 / max 200) + batch metadata + via `settings.list_by_key` (no N+1) — `projects.rs`, `repositories.rs`, + `projects_rpc.rs` +- [x] **P1** Index active projects + settings-by-key (migration `0005`) +- [x] **P3** Call `add_location` once on open (no duplicate upsert) + +### Carried from prior unmerged review work + +- [x] **P0** Cap RPC frame reads with chunked `fill_buf` (no unbounded + `read_until`) — `crates/retcon-core/src/frame.rs` +- [x] **P0** Cap Git stderr; kill child on stdout/stderr overflow — + `crates/retcon-git/src/lib.rs` +- [x] **P0** Bound event persist queue (`sync_channel` + `try_send`) — + `crates/retcon-core/src/event.rs` +- [x] **P1** Drain completed RPC `JoinSet` tasks; cap concurrent + connections (64) — `crates/retcon-core/src/server.rs` +- [x] **P1** Enforce outbound frame size (`MAX_FRAME_BYTES`) — + `server.rs` --- -## Completed schema work +## Still open + +### P1 + +- [ ] **P1** Dedicated job-persistence writer actor (replace sync DB writes + on job update/finish) — `crates/retcon-core/src/jobs.rs` +- [ ] **P1** Move event prune / lifecycle maintain / artifact cleanup off + async request & emit paths (`spawn_blocking` or writer-side prune) — + `event.rs`, `lifecycle.rs` +- [ ] **P1** Browser reader cleanup generation ID (don't clear a newly + started process) — `crates/retcon-core/src/spikes/browser.rs` +- [ ] **P1** Align Git 4MiB output cap with 1MiB protocol frame (page large + diffs as artifacts / lower per-method caps) — `retcon-git`, spikes/git +- [ ] **P1** Run `project.clone` as a supervised background job with + progress events (RPC currently waits for the full clone) + +### P2 + +- [ ] **P2** `jobs.list` SQL-backed ordered pagination (not in-memory + collect+sort) — `crates/retcon-core/src/jobs.rs` +- [ ] **P2** Paginate `SessionRepository::list_for_project` — + `crates/retcon-storage/src/repositories.rs` +- [ ] **P2** Artifact retention: stream/`EXISTS` instead of full hash set — + `crates/retcon-storage/src/artifacts.rs` +- [ ] **P2** `CREATE INDEX … ON agent_events(created_at)` + age prune off + emit path — migrations + `event.rs` +- [ ] **P2** Push `project.list` name/path filter into SQL (still filters + in Rust after the limited fetch) +- [ ] **P2** Wrap `add_location` upsert + project `updated_at` in one + transaction — `repositories.rs` +- [ ] **P2** Attach Playwright listeners to all pages (`context.on("page")`) + — `apps/browser-service/src/browser.ts` +- [ ] **P2** Floating panel instance IDs + `ValueKey` for stateful bodies — + `apps/desktop/lib/src/workspace.dart` +- [ ] **P2** Serialize/coalesce workspace layout persistence — + `apps/desktop/lib/src/workspace.dart` +- [ ] **P2** Fixed-index ring buffer for browser logs (vs `splice`) — + `apps/browser-service/src/browser.ts` +- [ ] **P2** BLOB-native fresh schema (skip TEXT→rebuild on new DBs) + +### P3 -- [x] **P3** UUID-as-BLOB schema migration (`retcon-storage` migration `0004`) - - Rebuilds v1-v3 databases transactionally, converts every UUID column to compact - 16-byte BLOBs, recreates indexes, and updates active repository/core bindings. +- [ ] **P3** Website asset generation mtime/hash short-circuit — + `apps/website/scripts/generate-brand-assets.mjs` +- [ ] **P3** Guard redundant focus `setState` in design-system controls — + `packages/retcon-design-system/lib/src/components/controls.dart` +- [ ] **P3** CI Playwright cache + path filters --- -## Suggested order of attack (completed) +## Suggested order of attack -1. Frame-size caps + event retention + job eviction (bounds) ✅ -2. Event writer off async path + terminal coalesce (throughput) ✅ -3. Recovery indexes + Git/browser evidence polish (latency) ✅ -4. Storage pragma / statement-cache tuning (incremental) ✅ +1. Job writer actor + prune/maintain off async paths +2. Browser process generation ID + Git/protocol size alignment +3. `project.clone` as a supervised job +4. Jobs/sessions pagination + event `created_at` index +5. Desktop panel instance IDs + save coalescing +6. Website/CI polish diff --git a/crates/retcon-core/src/event.rs b/crates/retcon-core/src/event.rs index d00f27d..3f7ca8d 100644 --- a/crates/retcon-core/src/event.rs +++ b/crates/retcon-core/src/event.rs @@ -4,7 +4,7 @@ use std::collections::VecDeque; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::mpsc::{self, Sender}; +use std::sync::mpsc::{self, SyncSender, TrySendError}; use std::sync::{Arc, Mutex}; use std::thread::{self, JoinHandle}; @@ -20,6 +20,7 @@ use retcon_storage::Database; const DEFAULT_MAX_EVENTS: usize = 10_000; const DEFAULT_MAX_AGE_DAYS: i64 = 7; const PRUNE_EVERY_N_INSERTS: u64 = 100; +const PERSIST_QUEUE_CAPACITY: usize = 2_048; #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -77,7 +78,7 @@ struct EventBusInner { log: Mutex, sender: broadcast::Sender>, database: Database, - persist_tx: Sender, + persist_tx: SyncSender, inserts_since_prune: AtomicU64, _writer: JoinHandle<()>, } @@ -100,7 +101,7 @@ impl EventBus { ) })?; let (sender, _) = broadcast::channel(1024); - let (persist_tx, persist_rx) = mpsc::channel(); + let (persist_tx, persist_rx) = mpsc::sync_channel(PERSIST_QUEUE_CAPACITY); let writer_db = database.clone(); let writer = thread::Builder::new() .name("event-writer".into()) @@ -173,21 +174,31 @@ impl EventBus { }; if durable { - self.inner - .persist_tx - .send(PersistJob { - event: Arc::clone(&event), - encoded_payload: encoded, - }) - .map_err(|error| internal_io(error.to_string()))?; - let count = self - .inner - .inserts_since_prune - .fetch_add(1, Ordering::Relaxed) - .saturating_add(1); - if count.is_multiple_of(PRUNE_EVERY_N_INSERTS) { - let _ = prune_database(&self.inner.database); - self.inner.inserts_since_prune.store(0, Ordering::Relaxed); + match self.inner.persist_tx.try_send(PersistJob { + event: Arc::clone(&event), + encoded_payload: encoded, + }) { + Ok(()) => { + let count = self + .inner + .inserts_since_prune + .fetch_add(1, Ordering::Relaxed) + .saturating_add(1); + if count.is_multiple_of(PRUNE_EVERY_N_INSERTS) { + let _ = prune_database(&self.inner.database); + self.inner.inserts_since_prune.store(0, Ordering::Relaxed); + } + } + Err(TrySendError::Full(_)) => { + tracing::warn!( + sequence = event.sequence, + kind = %event.kind, + "event persist queue full; dropping durable write" + ); + } + Err(TrySendError::Disconnected(_)) => { + return Err(internal_io("event writer disconnected".into())); + } } } diff --git a/crates/retcon-core/src/frame.rs b/crates/retcon-core/src/frame.rs index 4d90ad3..440fbdc 100644 --- a/crates/retcon-core/src/frame.rs +++ b/crates/retcon-core/src/frame.rs @@ -7,6 +7,9 @@ use retcon_protocol::MAX_FRAME_BYTES; /// Read one newline-delimited frame, rejecting payloads larger than [`MAX_FRAME_BYTES`]. /// +/// Reads in bounded chunks via `fill_buf` so a missing newline cannot force unbounded +/// allocation before the size check runs. +/// /// # Errors /// /// Returns [`CoreError`] when I/O fails or the frame exceeds the protocol limit. @@ -15,30 +18,45 @@ pub async fn read_capped_line( ) -> Result, CoreError> { let mut buffer = Vec::new(); loop { - let read = reader - .read_until(b'\n', &mut buffer) + let filled = reader + .fill_buf() .await .map_err(|error| CoreError::io("read RPC frame", error))?; - if read == 0 { + if filled.is_empty() { if buffer.is_empty() { return Ok(None); } return trimmed_line(buffer); } - if buffer.len() > MAX_FRAME_BYTES + 1 { - return Err(CoreError::new( - ErrorCode::InvalidRequest, - ErrorSource::Rpc, - "Retcon received an oversized local request.", - format!("frame exceeds {MAX_FRAME_BYTES} bytes"), - )); - } - if buffer.ends_with(b"\n") { + + if let Some(newline_at) = filled.iter().position(|&byte| byte == b'\n') { + let take = newline_at + 1; + if buffer.len().saturating_add(take) > MAX_FRAME_BYTES + 1 { + return oversized(); + } + buffer.extend_from_slice(&filled[..take]); + reader.consume(take); return trimmed_line(buffer); } + + let take = filled.len(); + if buffer.len().saturating_add(take) > MAX_FRAME_BYTES + 1 { + return oversized(); + } + buffer.extend_from_slice(filled); + reader.consume(take); } } +fn oversized() -> Result, CoreError> { + Err(CoreError::new( + ErrorCode::InvalidRequest, + ErrorSource::Rpc, + "Retcon received an oversized local request.", + format!("frame exceeds {MAX_FRAME_BYTES} bytes"), + )) +} + fn trimmed_line(mut buffer: Vec) -> Result, CoreError> { if buffer.ends_with(b"\n") { buffer.pop(); @@ -58,3 +76,29 @@ fn trimmed_line(mut buffer: Vec) -> Result, CoreError> { ) }) } + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::BufReader; + + #[tokio::test] + async fn rejects_oversize_frame_without_newline() { + let oversized = vec![b'a'; MAX_FRAME_BYTES + 2]; + let mut reader = BufReader::new(oversized.as_slice()); + let error = read_capped_line(&mut reader) + .await + .expect_err("must reject"); + assert!(error.technical_message.contains("frame exceeds")); + } + + #[tokio::test] + async fn reads_normal_frame() { + let mut reader = BufReader::new(b"{\"ok\":true}\n".as_slice()); + let line = read_capped_line(&mut reader) + .await + .expect("ok") + .expect("frame"); + assert_eq!(line, "{\"ok\":true}"); + } +} diff --git a/crates/retcon-core/src/lifecycle.rs b/crates/retcon-core/src/lifecycle.rs index afc18cd..fc905f0 100644 --- a/crates/retcon-core/src/lifecycle.rs +++ b/crates/retcon-core/src/lifecycle.rs @@ -273,7 +273,7 @@ mod tests { serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); assert_eq!(health["result"]["status"], "healthy"); assert_eq!(health["result"]["storage"]["status"], "healthy"); - assert_eq!(health["result"]["storage"]["schema_version"], 4); + assert_eq!(health["result"]["storage"]["schema_version"], 5); assert_eq!(shutdown["result"]["accepted"], true); runtime.wait_for_shutdown_signal().await.unwrap(); diff --git a/crates/retcon-core/src/projects.rs b/crates/retcon-core/src/projects.rs index b7d371d..4f19715 100644 --- a/crates/retcon-core/src/projects.rs +++ b/crates/retcon-core/src/projects.rs @@ -1,16 +1,17 @@ //! Project opening, lightweight repository analysis, and health reporting. -use std::collections::BTreeSet; +use std::collections::{BTreeSet, HashMap}; use std::path::{Path, PathBuf}; use serde_json::{Value, json}; -use tokio::process::Command; use uuid::Uuid; use crate::error::{CoreError, ErrorCode, ErrorSource}; use crate::state::CoreState; const METADATA_KEY: &str = "project.metadata"; +const DEFAULT_LIST_LIMIT: usize = 50; +const MAX_LIST_LIMIT: usize = 200; fn invalid(message: impl Into) -> CoreError { CoreError::new( @@ -45,12 +46,19 @@ fn canonical_directory(path: &str) -> Result { pub async fn open(state: &CoreState, path: &str) -> Result { let root = canonical_directory(path)?; let root_text = root.to_string_lossy().into_owned(); - let remote = git_output(&root, &["remote", "get-url", "origin"]) + let remote = retcon_git::run_git(&root, &["remote", "get-url", "origin"]) .await .ok() .filter(|s| !s.is_empty()); let project = match state.storage().projects().find_by_path(&root_text)? { - Some(project) => project, + Some(project) => { + // Refresh last-seen / remote without a duplicate create-path insert. + state + .storage() + .projects() + .add_location(project.id, &root_text, remote.as_deref())?; + project + } None => { let project = state .storage() @@ -67,11 +75,6 @@ pub async fn open(state: &CoreState, path: &str) -> Result { project } }; - // A reopened project may have gained an origin or moved between remotes. - state - .storage() - .projects() - .add_location(project.id, &root_text, remote.as_deref())?; let current = metadata(state, project.id)?; let defaults = json!({ "name": project.name, "repositoryPath": root_text, "remoteUrl": remote, @@ -81,13 +84,25 @@ pub async fn open(state: &CoreState, path: &str) -> Result { }); let merged = merge(defaults, current); save_metadata(state, project.id, &merged)?; + let analysis = tokio::task::spawn_blocking({ + let root = root.clone(); + move || analyze(&root) + }) + .await + .map_err(|error| { + CoreError::new( + ErrorCode::Internal, + ErrorSource::System, + "Retcon could not analyze that project.", + error.to_string(), + ) + })?; + let health = health(&root).await; state.emit( "project.opened", json!({"projectId": project.id, "path": root}), ); - Ok( - json!({"id": project.id, "metadata": merged, "analysis": analyze(&root), "health": health(&root).await}), - ) + Ok(json!({"id": project.id, "metadata": merged, "analysis": analysis, "health": health})) } /// Clone a repository and then open the resulting folder. @@ -103,39 +118,48 @@ pub async fn clone( if destination.exists() { return Err(invalid("clone destination already exists")); } - let output = Command::new("git") - .args(["clone", "--", remote_url, &destination.to_string_lossy()]) - .output() - .await - .map_err(|e| { - CoreError::new( - ErrorCode::Io, - ErrorSource::System, - "Git is required to clone a project.", - format!("start git clone: {e}"), - ) - })?; - if !output.status.success() { - return Err(CoreError::new( + let destination_text = destination.to_string_lossy().into_owned(); + retcon_git::run_git_cwd( + None, + &["clone", "--", remote_url, destination_text.as_str()], + ) + .await + .map_err(|error| { + CoreError::new( ErrorCode::Io, ErrorSource::System, "Retcon could not clone that repository.", - String::from_utf8_lossy(&output.stderr).trim().to_owned(), - )); - } - open(state, &destination.to_string_lossy()).await + error.stderr, + ) + })?; + open(state, &destination_text).await } /// List recent projects, optionally filtering by name or repository path. -pub fn list(state: &CoreState, query: Option<&str>) -> Result { +pub fn list( + state: &CoreState, + query: Option<&str>, + limit: Option, +) -> Result { let needle = query.unwrap_or_default().to_ascii_lowercase(); - let records = state + let limit = limit + .unwrap_or(DEFAULT_LIST_LIMIT) + .clamp(1, MAX_LIST_LIMIT); + let records = state.storage().projects().list(limit)?; + let metadata_by_scope = state .storage() - .projects() - .list()? + .settings() + .list_by_key(METADATA_KEY)? + .into_iter() + .map(|setting| (setting.scope, setting.value)) + .collect::>(); + let projects = records .into_iter() .filter_map(|project| { - let data = metadata(state, project.id).ok()?; + let data = metadata_by_scope + .get(&scope(project.id)) + .cloned() + .unwrap_or_else(|| json!({})); let name = data .get("name") .and_then(Value::as_str) @@ -150,7 +174,7 @@ pub fn list(state: &CoreState, query: Option<&str>) -> Result .then(|| json!({"id": project.id, "metadata": data, "updatedAt": project.updated_at})) }) .collect::>(); - Ok(json!({"projects": records})) + Ok(json!({"projects": projects})) } /// Persist a partial metadata update for a project. @@ -186,7 +210,20 @@ pub fn remove(state: &CoreState, id: Uuid) -> Result<(), CoreError> { /// Analyze a folder without adding it to the recent-project list. pub async fn inspect(path: &str) -> Result { let root = canonical_directory(path)?; - Ok(json!({"analysis": analyze(&root), "health": health(&root).await})) + let analysis = tokio::task::spawn_blocking({ + let root = root.clone(); + move || analyze(&root) + }) + .await + .map_err(|error| { + CoreError::new( + ErrorCode::Internal, + ErrorSource::System, + "Retcon could not analyze that project.", + error.to_string(), + ) + })?; + Ok(json!({"analysis": analysis, "health": health(&root).await})) } fn metadata(state: &CoreState, id: Uuid) -> Result { @@ -299,51 +336,48 @@ fn analyze(root: &Path) -> Value { } async fn health(root: &Path) -> Value { - let git_version = Command::new("git") - .arg("--version") - .output() + let git_version = retcon_git::run_git_cwd(None, &["--version"]) .await - .ok() - .filter(|o| o.status.success()) - .is_some(); - let git_status = git_output(root, &["status", "--porcelain=v1", "--branch"]) + .is_ok(); + let readable = retcon_git::run_git(root, &["rev-parse", "--is-inside-work-tree"]) .await - .ok(); - let branch = git_status - .as_deref() - .and_then(|s| s.lines().next()) - .and_then(|s| s.strip_prefix("## ")) - .map(|s| s.split("...").next().unwrap_or(s)); - let dirty = git_status - .as_deref() - .map(|s| s.lines().skip(1).next().is_some()) - .unwrap_or(false); + .ok() + .is_some_and(|value| value.trim() == "true"); + let branch = if readable { + retcon_git::run_git(root, &["rev-parse", "--abbrev-ref", "HEAD"]) + .await + .ok() + .filter(|value| !value.is_empty()) + } else { + None + }; + // Exit-code / empty-output probes avoid buffering full porcelain for large trees. + let dirty = if readable { + let staged_dirty = retcon_git::run_git(root, &["diff", "--cached", "--quiet"]) + .await + .is_err(); + let unstaged_dirty = retcon_git::run_git(root, &["diff", "--quiet"]).await.is_err(); + let untracked = retcon_git::run_git(root, &["ls-files", "--others", "--exclude-standard"]) + .await + .ok() + .is_some_and(|output| !output.trim().is_empty()); + staged_dirty || unstaged_dirty || untracked + } else { + false + }; let disk = std::fs::metadata(root) .ok() .map(|_| "available") .unwrap_or("unavailable"); json!({"checks": [ {"name":"Git installed", "status": if git_version {"pass"} else {"warning"}}, - {"name":"Repository readable", "status": if git_status.is_some() {"pass"} else {"warning"}}, + {"name":"Repository readable", "status": if readable {"pass"} else {"warning"}}, {"name":"Branch detected", "status": if branch.is_some() {"pass"} else {"warning"}, "detail": branch}, {"name":"Working tree", "status": if dirty {"warning"} else {"pass"}, "detail": if dirty {"uncommitted changes"} else {"clean"}}, {"name":"Project folder", "status": if disk == "available" {"pass"} else {"error"}}, {"name":"Environment template", "status": if root.join(".env.example").is_file() || !root.join(".env").exists() {"pass"} else {"warning"}, "detail":"Review local environment files before running commands."} ]}) } -async fn git_output(root: &Path, args: &[&str]) -> Result { - let output = Command::new("git") - .args(args) - .current_dir(root) - .output() - .await - .map_err(|_| ())?; - output - .status - .success() - .then(|| String::from_utf8_lossy(&output.stdout).trim().to_owned()) - .ok_or(()) -} #[cfg(test)] mod tests { @@ -379,7 +413,7 @@ mod tests { &json!({"pinned": true, "testCommand": "cargo test"}), ) .unwrap(); - let projects = list(&state, None).unwrap(); + let projects = list(&state, None, None).unwrap(); assert_eq!(projects["projects"][0]["metadata"]["pinned"], true); assert_eq!( projects["projects"][0]["metadata"]["testCommand"], diff --git a/crates/retcon-core/src/projects_rpc.rs b/crates/retcon-core/src/projects_rpc.rs index d04031c..fbf7b42 100644 --- a/crates/retcon-core/src/projects_rpc.rs +++ b/crates/retcon-core/src/projects_rpc.rs @@ -68,12 +68,19 @@ pub async fn handle(state: CoreState, request: Request) -> Response { .unwrap_or_else(|e| failed(id, e)), _ => missing(id, "remoteUrl or destination"), }, - "project.list" => projects::list( - &state, - params.get("query").and_then(serde_json::Value::as_str), - ) - .map(|v| Response::ok(id, v)) - .unwrap_or_else(|e| failed(id, e)), + "project.list" => { + let limit = params + .get("limit") + .and_then(serde_json::Value::as_u64) + .map(|value| value as usize); + projects::list( + &state, + params.get("query").and_then(serde_json::Value::as_str), + limit, + ) + .map(|v| Response::ok(id, v)) + .unwrap_or_else(|e| failed(id, e)) + } "project.inspect" => match params.get("path").and_then(serde_json::Value::as_str) { Some(path) => projects::inspect(path) .await diff --git a/crates/retcon-core/src/server.rs b/crates/retcon-core/src/server.rs index 641fee9..152dbd5 100644 --- a/crates/retcon-core/src/server.rs +++ b/crates/retcon-core/src/server.rs @@ -15,6 +15,8 @@ use crate::rpc::{Request, Response}; use crate::state::CoreState; use retcon_protocol::AuthLine; +const MAX_RPC_CONNECTIONS: usize = 64; + pub struct Server { listener: TcpListener, token: String, @@ -46,6 +48,7 @@ impl Server { heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { + let at_capacity = connections.len() >= MAX_RPC_CONNECTIONS; tokio::select! { _ = heartbeat.tick() => { let _ = self.state.events().emit_volatile( @@ -53,7 +56,8 @@ impl Server { json!({"uptime_ms": self.state.uptime().as_millis()}), ); }, - result = self.listener.accept() => match result { + Some(_) = connections.join_next(), if !connections.is_empty() => {} + result = self.listener.accept(), if !at_capacity => match result { Ok((stream, _)) => { let token = self.token.clone(); let state = self.state.clone(); @@ -126,6 +130,17 @@ async fn serve_connection( ) })?; encoded.push(b'\n'); + if encoded.len() > retcon_protocol::MAX_FRAME_BYTES + 1 { + return Err(CoreError::new( + ErrorCode::Internal, + ErrorSource::Rpc, + "Retcon could not send an oversized local response.", + format!( + "outbound frame exceeds {} bytes", + retcon_protocol::MAX_FRAME_BYTES + ), + )); + } writer .write_all(&encoded) .await diff --git a/crates/retcon-git/src/lib.rs b/crates/retcon-git/src/lib.rs index eae590d..d74b7ac 100644 --- a/crates/retcon-git/src/lib.rs +++ b/crates/retcon-git/src/lib.rs @@ -39,22 +39,40 @@ impl std::fmt::Display for GitError { /// /// Returns a [`GitError`] carrying Git's exit code and stderr on failure. pub async fn run_git(repo: &Path, args: &[&str]) -> Result { - run_git_limited(repo, args, MAX_GIT_OUTPUT_BYTES).await + run_git_cwd(Some(repo), args).await } -async fn run_git_limited(repo: &Path, args: &[&str], max_bytes: usize) -> Result { +/// Run `git` with `args` in an optional working directory, returning trimmed stdout. +/// +/// Use this for commands that do not require an existing repository (for example `clone`). +/// +/// # Errors +/// +/// Returns a [`GitError`] carrying Git's exit code and stderr on failure. +pub async fn run_git_cwd(cwd: Option<&Path>, args: &[&str]) -> Result { + run_git_limited(cwd, args, MAX_GIT_OUTPUT_BYTES).await +} + +async fn run_git_limited( + cwd: Option<&Path>, + args: &[&str], + max_bytes: usize, +) -> Result { let command_label = args.join(" "); - let mut child = Command::new("git") + let mut command = Command::new("git"); + command .args(args) - .current_dir(repo) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()) - .spawn() - .map_err(|e| GitError { - command: command_label.clone(), - exit_code: None, - stderr: format!("failed to launch git: {e}"), - })?; + .kill_on_drop(true); + if let Some(cwd) = cwd { + command.current_dir(cwd); + } + let mut child = command.spawn().map_err(|e| GitError { + command: command_label.clone(), + exit_code: None, + stderr: format!("failed to launch git: {e}"), + })?; let mut stdout = child.stdout.take().ok_or_else(|| GitError { command: command_label.clone(), @@ -67,66 +85,54 @@ async fn run_git_limited(repo: &Path, args: &[&str], max_bytes: usize) -> Result stderr: "git process has no stderr".into(), })?; - let stdout_task = tokio::spawn(async move { - let mut buffer = Vec::new(); - let mut chunk = [0_u8; 8192]; - loop { - let read = stdout - .read(&mut chunk) - .await - .map_err(|error| error.to_string())?; - if read == 0 { - break; + let stdout_fut = read_capped_stream(&mut stdout, max_bytes, "stdout"); + let stderr_fut = read_capped_stream(&mut stderr, max_bytes, "stderr"); + tokio::pin!(stdout_fut); + tokio::pin!(stderr_fut); + + let mut stdout_bytes: Option, String>> = None; + let mut stderr_bytes: Option, String>> = None; + + let status = loop { + tokio::select! { + result = &mut stdout_fut, if stdout_bytes.is_none() => { + if result.is_err() { + let _ = child.start_kill(); + } + stdout_bytes = Some(result); } - if buffer.len().saturating_add(read) > max_bytes { - return Err(format!("git output exceeded {max_bytes} bytes")); + result = &mut stderr_fut, if stderr_bytes.is_none() => { + if result.is_err() { + let _ = child.start_kill(); + } + stderr_bytes = Some(result); } - buffer.extend_from_slice(&chunk[..read]); - } - Ok(buffer) - }); - - let stderr_task = tokio::spawn(async move { - let mut buffer = Vec::new(); - let mut chunk = [0_u8; 1024]; - loop { - let read = stderr - .read(&mut chunk) - .await - .map_err(|error| error.to_string())?; - if read == 0 { - break; + status = child.wait() => { + break status.map_err(|e| GitError { + command: command_label.clone(), + exit_code: None, + stderr: format!("failed to wait for git: {e}"), + })?; } - buffer.extend_from_slice(&chunk[..read]); } - Ok(buffer) - }); + }; - let status = child.wait().await.map_err(|e| GitError { - command: command_label.clone(), - exit_code: None, - stderr: format!("failed to wait for git: {e}"), - })?; + if stdout_bytes.is_none() { + stdout_bytes = Some(stdout_fut.await); + } + if stderr_bytes.is_none() { + stderr_bytes = Some(stderr_fut.await); + } - let stdout_bytes = stdout_task - .await - .map_err(|e| GitError { - command: command_label.clone(), - exit_code: status.code(), - stderr: format!("stdout reader failed: {e}"), - })? + let stdout_bytes = stdout_bytes + .expect("stdout reader settled") .map_err(|message| GitError { command: command_label.clone(), exit_code: status.code(), stderr: message, })?; - let stderr_bytes = stderr_task - .await - .map_err(|e| GitError { - command: command_label.clone(), - exit_code: status.code(), - stderr: format!("stderr reader failed: {e}"), - })? + let stderr_bytes = stderr_bytes + .expect("stderr reader settled") .map_err(|message| GitError { command: command_label.clone(), exit_code: status.code(), @@ -144,6 +150,29 @@ async fn run_git_limited(repo: &Path, args: &[&str], max_bytes: usize) -> Result } } +async fn read_capped_stream( + stream: &mut (impl AsyncReadExt + Unpin), + max_bytes: usize, + label: &str, +) -> Result, String> { + let mut buffer = Vec::new(); + let mut chunk = [0_u8; 8192]; + loop { + let read = stream + .read(&mut chunk) + .await + .map_err(|error| error.to_string())?; + if read == 0 { + break; + } + if buffer.len().saturating_add(read) > max_bytes { + return Err(format!("git {label} exceeded {max_bytes} bytes")); + } + buffer.extend_from_slice(&chunk[..read]); + } + Ok(buffer) +} + /// One entry from `git status --porcelain`. #[derive(Debug, Clone, Serialize)] pub struct StatusEntry { diff --git a/crates/retcon-storage/src/database.rs b/crates/retcon-storage/src/database.rs index 626ecba..1b05a6f 100644 --- a/crates/retcon-storage/src/database.rs +++ b/crates/retcon-storage/src/database.rs @@ -630,7 +630,7 @@ mod tests { drop(connection); let upgraded = Database::open(path).unwrap(); - assert_eq!(upgraded.schema_version().unwrap(), 4); + assert_eq!(upgraded.schema_version().unwrap(), LATEST_VERSION); let job_id = Uuid::new_v4(); upgraded .execute( diff --git a/crates/retcon-storage/src/migrations.rs b/crates/retcon-storage/src/migrations.rs index d9cddcc..b8a4d2e 100644 --- a/crates/retcon-storage/src/migrations.rs +++ b/crates/retcon-storage/src/migrations.rs @@ -6,7 +6,7 @@ pub(crate) struct Migration { pub sql: &'static str, } -pub(crate) const LATEST_VERSION: u32 = 4; +pub(crate) const LATEST_VERSION: u32 = 5; pub(crate) const MIGRATIONS: &[Migration] = &[ Migration { @@ -29,4 +29,9 @@ pub(crate) const MIGRATIONS: &[Migration] = &[ name: "compact UUID storage", sql: include_str!("migrations/0004_uuid_blobs.sql"), }, + Migration { + version: 5, + name: "project list and settings indexes", + sql: include_str!("migrations/0005_project_list_indexes.sql"), + }, ]; diff --git a/crates/retcon-storage/src/migrations/0005_project_list_indexes.sql b/crates/retcon-storage/src/migrations/0005_project_list_indexes.sql new file mode 100644 index 0000000..ef11eed --- /dev/null +++ b/crates/retcon-storage/src/migrations/0005_project_list_indexes.sql @@ -0,0 +1,7 @@ +-- Active project listing ordered by recency. +CREATE INDEX IF NOT EXISTS idx_projects_active_updated + ON projects(updated_at DESC) + WHERE archived_at IS NULL; + +-- Project metadata hydration for project.list. +CREATE INDEX IF NOT EXISTS idx_settings_key ON settings(key); diff --git a/crates/retcon-storage/src/repositories.rs b/crates/retcon-storage/src/repositories.rs index 441fd45..db72e49 100644 --- a/crates/retcon-storage/src/repositories.rs +++ b/crates/retcon-storage/src/repositories.rs @@ -198,12 +198,14 @@ impl ProjectRepository<'_> { }) } /// List active projects, most recently updated first. - pub fn list(&self) -> Result> { + pub fn list(&self, limit: usize) -> Result> { self.0.read(|db| { let mut statement = db.prepare( - "SELECT id,name,created_at,updated_at FROM projects WHERE archived_at IS NULL ORDER BY updated_at DESC", + "SELECT id,name,created_at,updated_at FROM projects WHERE archived_at IS NULL ORDER BY updated_at DESC LIMIT ?1", )?; - statement.query_map([], row_project)?.collect() + statement + .query_map([limit as i64], row_project)? + .collect() }) } @@ -385,6 +387,33 @@ impl SettingsRepository<'_> { .optional() }) } + + /// Load every setting row for a key (used to hydrate project metadata in one query). + pub fn list_by_key(&self, key: &str) -> Result> { + self.0.read(|db| { + let mut statement = db.prepare( + "SELECT scope,key,value_json,updated_at FROM settings WHERE key=?1", + )?; + statement + .query_map([key], |row| { + let encoded: String = row.get(2)?; + let value = serde_json::from_str(&encoded).map_err(|error| { + rusqlite::Error::FromSqlConversionFailure( + encoded.len(), + rusqlite::types::Type::Text, + Box::new(error), + ) + })?; + Ok(Setting { + scope: row.get(0)?, + key: row.get(1)?, + value, + updated_at: row.get(3)?, + }) + })? + .collect() + }) + } pub fn delete(&self, scope: &str, key: &str) -> Result { Ok(self.0.execute( "DELETE FROM settings WHERE scope=?1 AND key=?2",