diff --git a/src/commands/remote/history.rs b/src/commands/remote/history.rs index ee773f0..1065482 100644 --- a/src/commands/remote/history.rs +++ b/src/commands/remote/history.rs @@ -44,8 +44,14 @@ pub fn push_history_to_remote(project: Option<&str>, remote_name: &str) -> Resul &project.repos, )?; } - let pushed = - push_project_history_events(remote, &token, &remote_project.slug, &root, &project_id)?; + let pushed = push_project_history_events( + remote, + &token, + &remote_project.slug, + &root, + &project_id, + remote_name, + )?; println!( "{} {} {}", out::movement("pushed history"), @@ -77,25 +83,67 @@ pub(super) fn push_project_history_events( project_slug: &str, root: &Path, project_id: &str, + remote_name: &str, ) -> Result { refresh_project_history(root, project_id)?; let events = load_history_events(root, project_id)?; if events.is_empty() { return Ok(0); } + let encoded = events + .iter() + .map(|event| serde_json::to_string(event).context("failed to encode history event")) + .collect::>>()?; + let state = load_history_sync_state(root, project_id)?; + let plan = plan_history_push(&encoded, state.get(remote_name)); + let to_send: &[HistoryEvent] = match plan { + HistoryPushPlan::UpToDate => { + record_history_sync(root, project_id, remote_name, &encoded, state)?; + return Ok(events.len()); + } + HistoryPushPlan::Tail(from) => &events[from..], + HistoryPushPlan::Full => &events, + }; + + let batches: Vec<&[HistoryEvent]> = to_send.chunks(HISTORY_PAGE_SIZE).collect(); + if batches.len() > 1 { + println!( + "{}", + out::muted(format!( + "syncing history to {remote_name}: {} event(s) in {} request(s)…", + to_send.len(), + batches.len() + )) + ); + } // Batched so a project ledger of thousands of events never rides in one - // request body; each batch upserts independently and is idempotent. + // request body; each batch upserts independently and is idempotent, so + // the batches go out concurrently — a full push of a large ledger is + // bounded by the slowest request, not their sum. + let path = format!("/projects/{project_slug}/history-events"); + let outcomes: Vec> = std::thread::scope(|scope| { + let mut handles = Vec::new(); + for group in batches.chunks(HISTORY_PUSH_CONCURRENCY) { + let started: Vec<_> = group + .iter() + .map(|batch| { + let path = path.as_str(); + scope.spawn(move || { + let payload = json!({ "events": batch }); + request_json::(remote, token, "POST", path, Some(&payload)) + }) + }) + .collect(); + for handle in started { + handles.push(handle.join().unwrap_or_else(|_| Err(anyhow::anyhow!("history push thread panicked")))); + } + } + handles + }); let mut accepted = 0; let mut failed = 0; - for batch in events.chunks(HISTORY_PAGE_SIZE) { - let payload = json!({ "events": batch }); - let response: RemoteHistoryPush = request_json( - remote, - token, - "POST", - &format!("/projects/{project_slug}/history-events"), - Some(&payload), - )?; + for outcome in outcomes { + let response = outcome?; accepted += response.inserted_count + response.updated_count + response.skipped_count; failed += response.failed_count; } @@ -104,10 +152,118 @@ pub(super) fn push_project_history_events( "{} {failed} history event(s) were rejected by the sync remote and are missing there; the next push retries them", out::warn("warning:") ); + } else { + // Only a fully accepted push moves the cursor: a rejected event stays + // ahead of it and rides again next time. + record_history_sync(root, project_id, remote_name, &encoded, state)?; + } + if let HistoryPushPlan::Tail(from) = plan { + // What the remote holds now, for the "N event(s) synced" line. + accepted += from; } Ok(accepted) } +/// How many history requests are in flight at once during a push. +const HISTORY_PUSH_CONCURRENCY: usize = 4; + +/// What one push has to send, given what the remote already holds. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum HistoryPushPlan { + /// The ledger is byte-for-byte what was last pushed: nothing to send. + UpToDate, + /// The ledger only grew since the last push: send from this index on. + Tail(usize), + /// An earlier event changed (a rebuild enriched it) or nothing was ever + /// pushed: send everything. + Full, +} + +/// Per-remote memory of what a push last sent: how many events, and a +/// fingerprint of exactly those encoded lines. A later push compares the +/// ledger's prefix against it; an append-only ledger (the normal case) then +/// costs one request for the new events, and an unchanged one costs none. +#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub(super) struct HistorySyncCursor { + pub event_count: usize, + pub fingerprint: String, +} + +pub(super) type HistorySyncState = std::collections::BTreeMap; + +/// FNV-1a over the encoded lines, in order. A content fingerprint, not a +/// security boundary; it is deliberately self-contained so it never changes +/// underneath a stored cursor. +pub(super) fn history_fingerprint(encoded_lines: &[String]) -> String { + let mut hash: u64 = 0xcbf2_9ce4_8422_2325; + for line in encoded_lines { + for byte in line.bytes().chain(std::iter::once(b'\n')) { + hash ^= u64::from(byte); + hash = hash.wrapping_mul(0x0000_0100_0000_01b3); + } + } + format!("fnv1a64:{hash:016x}") +} + +pub(super) fn plan_history_push( + encoded_lines: &[String], + cursor: Option<&HistorySyncCursor>, +) -> HistoryPushPlan { + let Some(cursor) = cursor else { return HistoryPushPlan::Full }; + if cursor.event_count == 0 || cursor.event_count > encoded_lines.len() { + return HistoryPushPlan::Full; + } + if history_fingerprint(&encoded_lines[..cursor.event_count]) != cursor.fingerprint { + return HistoryPushPlan::Full; + } + if cursor.event_count == encoded_lines.len() { + HistoryPushPlan::UpToDate + } else { + HistoryPushPlan::Tail(cursor.event_count) + } +} + +fn history_sync_state_path(root: &Path, project_id: &str) -> std::path::PathBuf { + crate::store::history_path(root, project_id).with_file_name(format!("{project_id}.history-sync.json")) +} + +pub(super) fn load_history_sync_state(root: &Path, project_id: &str) -> Result { + let path = history_sync_state_path(root, project_id); + if !path.exists() { + return Ok(HistorySyncState::new()); + } + let text = std::fs::read_to_string(&path) + .with_context(|| format!("failed to read {}", path.display()))?; + // A cursor that cannot be read is a cursor that never existed: the next + // push is a full one, which is always correct. + Ok(serde_json::from_str(&text).unwrap_or_default()) +} + +fn record_history_sync( + root: &Path, + project_id: &str, + remote_name: &str, + encoded_lines: &[String], + mut state: HistorySyncState, +) -> Result<()> { + state.insert( + remote_name.to_string(), + HistorySyncCursor { + event_count: encoded_lines.len(), + fingerprint: history_fingerprint(encoded_lines), + }, + ); + let path = history_sync_state_path(root, project_id); + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("failed to create {}", parent.display()))?; + } + let body = serde_json::to_string_pretty(&state).context("failed to encode history sync state")?; + std::fs::write(&path, format!("{body}\n")) + .with_context(|| format!("failed to write {}", path.display())) +} + /// How many history events ride in one request, both directions. const HISTORY_PAGE_SIZE: usize = 500; @@ -166,3 +322,35 @@ pub(super) fn fetch_project_history_events( Ok(super::decode_history_events(&all, project_identifier)) } + +#[cfg(test)] +mod push_plan_tests { + use super::*; + + fn lines(count: usize) -> Vec { + (0..count).map(|index| format!("{{\"eventId\":\"e{index}\"}}")).collect() + } + + #[test] + fn nothing_recorded_means_a_full_push() { + assert_eq!(plan_history_push(&lines(3), None), HistoryPushPlan::Full); + } + + #[test] + fn an_unchanged_ledger_sends_nothing_and_an_appended_one_sends_the_tail() { + let pushed = lines(3); + let cursor = HistorySyncCursor { event_count: 3, fingerprint: history_fingerprint(&pushed) }; + assert_eq!(plan_history_push(&pushed, Some(&cursor)), HistoryPushPlan::UpToDate); + assert_eq!(plan_history_push(&lines(5), Some(&cursor)), HistoryPushPlan::Tail(3)); + } + + #[test] + fn a_rewritten_prefix_or_a_shrunken_ledger_sends_everything() { + let pushed = lines(3); + let cursor = HistorySyncCursor { event_count: 3, fingerprint: history_fingerprint(&pushed) }; + let mut rewritten = lines(4); + rewritten[1] = "{\"eventId\":\"e1\",\"message\":\"enriched\"}".to_string(); + assert_eq!(plan_history_push(&rewritten, Some(&cursor)), HistoryPushPlan::Full); + assert_eq!(plan_history_push(&lines(2), Some(&cursor)), HistoryPushPlan::Full); + } +} diff --git a/src/commands/remote/push.rs b/src/commands/remote/push.rs index 959ac0c..654ca97 100644 --- a/src/commands/remote/push.rs +++ b/src/commands/remote/push.rs @@ -678,6 +678,9 @@ fn push_active_bundle_to_remote( .or_else(|| active.bundle.project_id.clone()) .or_else(|| config.active_project.clone()) .context("No project selected. Pass --project or run `knit init `.")?; + // Say that the sync has started: the branch pushes above finish in + // seconds, and a silent minute after them read as a hang. + println!("{}", out::muted(format!("syncing {} to {remote_name}…", active.bundle.id))); let remote = resolve_remote(&config, remote_name)?; let token = resolve_token(remote_name, remote)?; let local_project = load_project_if_present(&active.root, &project_id)?; @@ -701,6 +704,7 @@ fn push_active_bundle_to_remote( &pushed_project.slug, &active.root, &project_id, + remote_name, ); println!( diff --git a/src/store.rs b/src/store.rs index 67854b5..d751393 100644 --- a/src/store.rs +++ b/src/store.rs @@ -508,7 +508,21 @@ fn resolve_bundle_id( return Ok((bundle_id.clone(), BundleResolutionSource::Config)); } - bail!("No active Knit bundle found. Run `knit bundle \"feature title\"` first.") + // No fallback is set. Whether that is "make a bundle" or "say which one" + // depends on what exists: telling a person with four open bundles to + // create one is wrong advice. + let open_bundles = open_bundle_ids(root)?; + match open_bundles.as_slice() { + [] => bail!("No active Knit bundle found. Run `knit bundle \"feature title\"` first."), + [only] => bail!( + "No active Knit bundle is selected. Bundle `{only}` is open: run the same command with `--bundle {only}`, run it from `.knit/worktrees/{only}//`, or select it with `knit switch {only} --workspace`." + ), + many => bail!( + "No active Knit bundle is selected and {} bundles are open: {}. Run the same command with `--bundle `, or run it from `.knit/worktrees///`.", + many.len(), + many.join(", ") + ), + } } pub fn ensure_workspace_fallback_status_is_unambiguous(active: &ActiveBundle) -> Result<()> { diff --git a/tests/bundle.rs b/tests/bundle.rs index c277d37..a7d5998 100644 --- a/tests/bundle.rs +++ b/tests/bundle.rs @@ -645,3 +645,48 @@ fn bundle_add_records_repo_before_materializing_worktree() { fs::remove_dir_all(root).unwrap(); } + +#[test] +fn no_workspace_fallback_names_the_open_bundles_instead_of_asking_for_a_new_one() { + let root = unique_temp_dir(); + let backend = root.join("backend"); + let workspace = root.join("workspace"); + fs::create_dir_all(&workspace).unwrap(); + init_repo(&backend, "backend"); + + knit(&workspace, ["bundle", "fix a"]); + knit(&workspace, ["bundle", "add", backend.to_str().unwrap()]); + + // No fallback selected, one bundle open: say which one and how to pick it. + let config_path = workspace.join(".knit/config.json"); + let clear_fallback = || { + let mut config: Value = + serde_json::from_str(&fs::read_to_string(&config_path).unwrap()).unwrap(); + config.as_object_mut().unwrap().remove("activeBundle"); + fs::write(&config_path, format!("{}\n", serde_json::to_string_pretty(&config).unwrap())) + .unwrap(); + }; + clear_fallback(); + let one = knit_fails(&workspace, ["status"]); + assert!(one.contains("Bundle `fix-a` is open"), "{one}"); + assert!(one.contains("--bundle fix-a"), "{one}"); + assert!(!one.contains("feature title"), "told to create a bundle while one is open: {one}"); + + // Several open: list them all, ask for --bundle, never "create one". + knit(&workspace, ["bundle", "fix b"]); + knit(&workspace, ["bundle", "add", backend.to_str().unwrap()]); + clear_fallback(); + let many = knit_fails(&workspace, ["publish", "create"]); + assert!(many.contains("2 bundles are open: fix-a, fix-b"), "{many}"); + assert!(many.contains("--bundle "), "{many}"); + assert!(!many.contains("feature title"), "{many}"); + + // An empty workspace still gets the original advice. + let empty = root.join("empty"); + fs::create_dir_all(&empty).unwrap(); + knit(&empty, ["init", "demo"]); + let none = knit_fails(&empty, ["status"]); + assert!(none.contains("Run `knit bundle \"feature title\"` first"), "{none}"); + + fs::remove_dir_all(root).unwrap(); +} diff --git a/tests/common/mod.rs b/tests/common/mod.rs index f9b2c22..4eac882 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -1691,10 +1691,21 @@ fn handle_fake_remote_push_request( fs::write(&record, existing).unwrap(); (200, format!("{{\"data\":{{\"id\":\"{repo_id}\"}}}}")) } - ("POST", ["api", "v1", "projects", _, "history-events"]) => ( - 201, - "{\"data\":{\"insertedCount\":0,\"skippedCount\":0}}".to_string(), - ), + ("POST", ["api", "v1", "projects", _, "history-events"]) => { + // Every history push is recorded, one line per request, so a + // test can see how many requests a sync made and which events + // rode in each. + let record = dir.join("history-pushes.jsonl"); + let mut existing = fs::read_to_string(&record).unwrap_or_default(); + existing.push_str(&body.to_string()); + existing.push('\n'); + fs::write(&record, existing).unwrap(); + let count = body["events"].as_array().map(Vec::len).unwrap_or(0); + ( + 201, + format!("{{\"data\":{{\"insertedCount\":{count},\"skippedCount\":0}}}}"), + ) + } ("POST", ["api", "v1", "projects", _, "bundles"]) => { let slug = body["slug"].as_str().unwrap_or("unknown").to_string(); ( @@ -1945,3 +1956,25 @@ fn handle_fake_remote_request(stream: &mut std::net::TcpStream, dir: &Path) -> s )?; stream.flush() } + +/// The history pushes the fake push remote received: one entry per request, +/// each the list of event ids that request carried. +pub fn recorded_history_pushes(dir: &Path) -> Vec> { + fs::read_to_string(dir.join("history-pushes.jsonl")) + .unwrap_or_default() + .lines() + .filter(|line| !line.trim().is_empty()) + .map(|line| { + let body: serde_json::Value = serde_json::from_str(line).unwrap(); + body["events"] + .as_array() + .map(|events| { + events + .iter() + .filter_map(|event| event["eventId"].as_str().map(str::to_string)) + .collect() + }) + .unwrap_or_default() + }) + .collect() +} diff --git a/tests/sync.rs b/tests/sync.rs index 4801c76..556469a 100644 --- a/tests/sync.rs +++ b/tests/sync.rs @@ -2367,3 +2367,70 @@ fn pull_reconcile_reports_forge_missing_repos_honestly() { fs::remove_dir_all(root).unwrap(); } + +#[test] +fn sync_push_history_sends_only_what_the_remote_does_not_have_yet() { + let root = unique_temp_dir(); + let (_remote, backend, _collaborator) = init_remote_repo(&root, "backend"); + let workspace = root.join("workspace"); + fs::create_dir_all(&workspace).unwrap(); + + knit(&workspace, ["init", "demo"]); + knit( + &workspace, + ["project", "add", "backend", backend.to_str().unwrap()], + ); + let fake_dir = root.join("fake-remote"); + let base_url = spawn_fake_remote_push_api(&fake_dir); + knit(&workspace, ["remote", "add", "hosted", &base_url]); + let env = [("KNIT_REMOTE_TOKEN", "owner-token")]; + + knit(&workspace, ["bundle", "ledger work", "--repo", "backend"]); + let feature = workspace.join(".knit/worktrees/ledger-work/backend"); + append_line(&feature.join("app.txt"), "first"); + knit(&workspace, ["commit", "--all", "-m", "First"]); + + // The first push sends the whole ledger. + let first = knit_with_env(&workspace, ["sync", "push", "--history"], &env); + assert!(first.contains("pushed history"), "{first}"); + let pushes = common::recorded_history_pushes(&fake_dir); + assert_eq!(pushes.len(), 1, "{pushes:?}"); + let initial: Vec = pushes[0].clone(); + assert!(!initial.is_empty()); + + // Nothing changed: the second push makes no history request at all, and + // still reports what the remote holds. + let second = knit_with_env(&workspace, ["sync", "push", "--history"], &env); + assert!(second.contains(&format!("{} event(s)", initial.len())), "{second}"); + assert_eq!(common::recorded_history_pushes(&fake_dir).len(), 1, "an unchanged ledger was pushed again"); + + // One more commit: only the events it added ride in the third push. + append_line(&feature.join("app.txt"), "second"); + knit(&workspace, ["commit", "--all", "-m", "Second"]); + let third = knit_with_env(&workspace, ["sync", "push", "--history"], &env); + let pushes = common::recorded_history_pushes(&fake_dir); + assert_eq!(pushes.len(), 2, "{pushes:?}"); + assert!(!pushes[1].is_empty(), "the new commit produced no event"); + assert!( + pushes[1].iter().all(|id| !initial.contains(id)), + "already-synced events rode again: {:?}", + pushes[1] + ); + assert!( + third.contains(&format!("{} event(s)", initial.len() + pushes[1].len())), + "{third}" + ); + + // A second remote starts from nothing: it gets the whole ledger, and the + // first remote's cursor is untouched. + let other_dir = root.join("other-remote"); + let other_url = spawn_fake_remote_push_api(&other_dir); + knit(&workspace, ["remote", "add", "mirror", &other_url]); + knit_with_env(&workspace, ["sync", "push", "--history", "--remote", "mirror"], &env); + let mirror = common::recorded_history_pushes(&other_dir); + assert_eq!(mirror.len(), 1, "{mirror:?}"); + assert_eq!(mirror[0].len(), initial.len() + pushes[1].len()); + assert_eq!(common::recorded_history_pushes(&fake_dir).len(), 2); + + fs::remove_dir_all(root).unwrap(); +}