push history incremental and honest bundle errors (knit)

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<usize> {
     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::<Result<Vec<_>>>()?;
+    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<Result<RemoteHistoryPush>> = 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::<RemoteHistoryPush>(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<String, HistorySyncCursor>;
+
+/// 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<HistorySyncState> {
+    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<String> {
+        (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);
+    }
+}

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 <name>`.")?;
+    // 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!(

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}/<repo>/`, 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 <bundle>`, or run it from `.knit/worktrees/<bundle>/<repo>/`.",
+            many.len(),
+            many.join(", ")
+        ),
+    }
 }
 
 pub fn ensure_workspace_fallback_status_is_unambiguous(active: &ActiveBundle) -> Result<()> {

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 <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();
+}

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<Vec<String>> {
+    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()
+}