From 28a0a5a2bc454155a5d1c823344474be2c3f1f19 Mon Sep 17 00:00:00 2001
From: Ragnor Comerford <hello@ragnor.co>
Date: Wed, 30 Sep 2026 16:46:42 +0200
Subject: [PATCH 01/11] test: pin merge concurrency across independent and
 shared targets

Merges into independent targets overlap: with one merge parked after its
table effects and before its manifest commit, a merge between two other
branches runs to completion. Merges into one target stay serialized: the
second waits for the first's target gate and then merges on top of it.
---
 crates/omnigraph/tests/failpoints.rs | 119 +++++++++++++++++++++++++++
 1 file changed, 119 insertions(+)

diff --git a/crates/omnigraph/tests/failpoints.rs b/crates/omnigraph/tests/failpoints.rs
index f881666f2..e82aae507 100644
--- a/crates/omnigraph/tests/failpoints.rs
+++ b/crates/omnigraph/tests/failpoints.rs
@@ -1965,6 +1965,125 @@ async fn cross_branch_writers_overlap_inside_schema_gate() {
     );
 }
 
+/// Create `source` and `target` from main and give each its own insert, so a
+/// merge between them is three-way and reaches its manifest commit.
+async fn diverged_merge_pair(db: &helpers::Session, source: &str, target: &str) {
+    db.branch_create(source).await.unwrap();
+    db.branch_create(target).await.unwrap();
+    for (branch, age) in [(source, 51), (target, 52)] {
+        db.mutate(
+            branch,
+            MUTATION_QUERIES,
+            "insert_person",
+            &mixed_params(&[("$name", &format!("On-{branch}"))], &[("$age", age)]),
+        )
+        .await
+        .unwrap();
+    }
+}
+
+/// Issue 643: merges into independent targets overlap. Merge A parks after
+/// its table effects, before its manifest commit, holding the shared schema
+/// permit and its own source and target branch gates; merge B between two
+/// other branches must run to completion meanwhile.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn independent_target_merges_overlap_issue_643() {
+    let _scenario = FailScenario::setup();
+    let dir = tempfile::tempdir().unwrap();
+    let db = helpers::init_and_load(&dir).await;
+    diverged_merge_pair(&db, "s1", "t1").await;
+    diverged_merge_pair(&db, "s2", "t2").await;
+    let db = std::sync::Arc::new(db);
+
+    let parked = helpers::failpoint::Rendezvous::park_first(
+        &catalog::BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT,
+    );
+    let merge_a_db = std::sync::Arc::clone(&db);
+    let merge_a = tokio::spawn(async move { merge_a_db.branch_merge("s1", "t1").await });
+    parked.wait_until_reached().await;
+
+    let merge_b_db = std::sync::Arc::clone(&db);
+    let merge_b = tokio::spawn(async move { merge_b_db.branch_merge("s2", "t2").await });
+    let merged_b = tokio::time::timeout(std::time::Duration::from_secs(10), merge_b)
+        .await
+        .expect("a merge into another target must complete while A is parked")
+        .unwrap()
+        .expect("B must merge");
+    assert_eq!(merged_b, omnigraph::db::MergeOutcome::Merged);
+    assert!(
+        !merge_a.is_finished(),
+        "A must still be parked before its manifest commit while B published"
+    );
+
+    parked.release();
+    assert_eq!(
+        merge_a.await.unwrap().expect("A must merge after release"),
+        omnigraph::db::MergeOutcome::Merged
+    );
+    for target in ["t1", "t2"] {
+        assert_eq!(
+            helpers::count_rows_branch(&db, target, "node:Person").await,
+            6,
+            "{target} holds the seed rows, its own insert and its source's"
+        );
+    }
+}
+
+/// Issue 643: merges into the same target stay serialized. With merge A
+/// parked before its manifest commit, holding the target's branch gate, a
+/// second merge into that target must wait; once A publishes, B merges on
+/// top of it.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn same_target_merges_serialize_issue_643() {
+    let _scenario = FailScenario::setup();
+    let dir = tempfile::tempdir().unwrap();
+    let db = helpers::init_and_load(&dir).await;
+    diverged_merge_pair(&db, "s1", "t").await;
+    db.branch_create("s2").await.unwrap();
+    db.mutate(
+        "s2",
+        MUTATION_QUERIES,
+        "insert_person",
+        &mixed_params(&[("$name", "On-s2")], &[("$age", 53)]),
+    )
+    .await
+    .unwrap();
+    let db = std::sync::Arc::new(db);
+
+    let parked = helpers::failpoint::Rendezvous::park_first(
+        &catalog::BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT,
+    );
+    let merge_a_db = std::sync::Arc::clone(&db);
+    let merge_a = tokio::spawn(async move { merge_a_db.branch_merge("s1", "t").await });
+    parked.wait_until_reached().await;
+
+    let merge_b_db = std::sync::Arc::clone(&db);
+    let mut merge_b = tokio::spawn(async move { merge_b_db.branch_merge("s2", "t").await });
+    assert!(
+        tokio::time::timeout(std::time::Duration::from_millis(500), &mut merge_b)
+            .await
+            .is_err(),
+        "a merge into the same target must wait while A holds that target's gate"
+    );
+
+    parked.release();
+    assert_eq!(
+        merge_a.await.unwrap().expect("A must merge after release"),
+        omnigraph::db::MergeOutcome::Merged
+    );
+    assert_eq!(
+        merge_b.await.unwrap().expect("B must merge after A"),
+        omnigraph::db::MergeOutcome::Merged
+    );
+    assert_eq!(
+        helpers::count_rows_branch(&db, "t", "node:Person").await,
+        7,
+        "t holds the seed rows, its own insert and both sources' inserts"
+    );
+}
+
 /// Branch create takes the schema gate's EXCLUSIVE side: no CAS covers its
 /// namespace inventory, so a sibling create with disjoint branch gates must
 /// wait at the gate, then refuse on the collision the first leaves behind.

From 636a1c5450ad872fe90706d0569687eed518fa92 Mon Sep 17 00:00:00 2001
From: Ragnor Comerford <hello@ragnor.co>
Date: Wed, 30 Sep 2026 16:51:54 +0200
Subject: [PATCH 02/11] test: a parked merge holds its shared permit through
 publication

The merge's counterpart of parked_writer_blocks_schema_apply: a schema apply
arriving while a merge is parked before its manifest commit must not answer
until the merge publishes, since past the exclusive gate it refuses at once
on the merge's source branch. The merge's result is intact afterwards.
---
 crates/omnigraph/tests/schema_apply.rs | 67 ++++++++++++++++++++++++++
 1 file changed, 67 insertions(+)

diff --git a/crates/omnigraph/tests/schema_apply.rs b/crates/omnigraph/tests/schema_apply.rs
index 6bb99ea74..b41fb2c01 100644
--- a/crates/omnigraph/tests/schema_apply.rs
+++ b/crates/omnigraph/tests/schema_apply.rs
@@ -455,6 +455,73 @@ async fn parked_writer_blocks_schema_apply() {
     assert_eq!(count_rows(&db, "node:Person").await, 5);
 }
 
+/// Issue 643, schema-apply boundary, and the merge's tripwire (the writer's
+/// is `parked_writer_blocks_schema_apply`): a merge parked before its
+/// manifest commit holds its SHARED schema permit, so a schema apply waits at
+/// the exclusive side. Past the gate the apply refuses at once, because the
+/// merge's source branch exists, so an apply that returns while the merge is
+/// parked means the merge let its permit go early. After the merge publishes,
+/// the apply refuses and the merge's result is intact.
+#[cfg(feature = "failpoints")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial_test::serial]
+async fn parked_merge_blocks_schema_apply_issue_643() {
+    use omnigraph::seams::catalog;
+
+    let dir = tempfile::tempdir().unwrap();
+    let db = init_and_load(&dir).await;
+    db.branch_create("s").await.unwrap();
+    for (branch, name, age) in [("s", "On-s", 61), ("main", "On-main", 62)] {
+        db.mutate(
+            branch,
+            MUTATION_QUERIES,
+            "insert_person",
+            &mixed_params(&[("$name", name)], &[("$age", age)]),
+        )
+        .await
+        .unwrap();
+    }
+    let db = Arc::new(db);
+    let desired = TEST_SCHEMA.replace("    age: I32?\n}", "    age: I32?\n    motto: String?\n}");
+
+    let parked = helpers::failpoint::Rendezvous::park_first(
+        &catalog::BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT,
+    );
+    let merge_db = Arc::clone(&db);
+    let merge = tokio::spawn(async move { merge_db.branch_merge("s", "main").await });
+    parked.wait_until_reached().await;
+
+    let schema_db = Arc::clone(&db);
+    let mut schema_task = tokio::spawn(async move { schema_db.apply_schema(&desired).await });
+    assert!(
+        tokio::time::timeout(std::time::Duration::from_secs(1), &mut schema_task)
+            .await
+            .is_err(),
+        "schema apply must wait behind a merge's held shared schema permit; it \
+         answered while the merge was parked",
+    );
+
+    parked.release();
+    assert_eq!(
+        merge
+            .await
+            .unwrap()
+            .expect("the merge must publish after release"),
+        omnigraph::db::MergeOutcome::Merged
+    );
+    let refusal = schema_task
+        .await
+        .unwrap()
+        .expect_err("schema apply refuses while the merge's source branch exists");
+    assert!(
+        refusal
+            .to_string()
+            .contains("schema apply requires a graph with only main"),
+        "{refusal}"
+    );
+    assert_eq!(count_rows(&db, "node:Person").await, 6);
+}
+
 /// ReadOnly opens participate in the process-local schema publication gate even
 /// though they perform no recovery writes. Park an open immediately before its
 /// source/IR/state read: schema apply must not reach staging until that coherent

From e4f92907a4b7401ca0b6fbc1c50276287ae461f1 Mon Sep 17 00:00:00 2001
From: Ragnor Comerford <hello@ragnor.co>
Date: Wed, 30 Sep 2026 16:55:52 +0200
Subject: [PATCH 03/11] test: pin branch deletes and same-source merges around
 a merge

Deleting a merge's target or source while the merge is parked before its
manifest commit waits on that branch's gate; the merge publishes and the
delete succeeds after it. Two merges from one source into two targets both
land with the source's row; whether they may overlap is left unpinned.
---
 crates/omnigraph/tests/failpoints.rs | 91 ++++++++++++++++++++++++++++
 1 file changed, 91 insertions(+)

diff --git a/crates/omnigraph/tests/failpoints.rs b/crates/omnigraph/tests/failpoints.rs
index e82aae507..90cbe91db 100644
--- a/crates/omnigraph/tests/failpoints.rs
+++ b/crates/omnigraph/tests/failpoints.rs
@@ -2084,6 +2084,97 @@ async fn same_target_merges_serialize_issue_643() {
     );
 }
 
+/// Issue 643, delete boundary: deleting a merge's target or source while
+/// the merge is parked before its manifest commit waits on that branch's
+/// gate. The merge then publishes, and the delete succeeds after it.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn merge_endpoint_delete_waits_for_the_merge_issue_643() {
+    for victim in ["t", "s"] {
+        let _scenario = FailScenario::setup();
+        let dir = tempfile::tempdir().unwrap();
+        let db = helpers::init_and_load(&dir).await;
+        diverged_merge_pair(&db, "s", "t").await;
+        let db = std::sync::Arc::new(db);
+
+        let parked = helpers::failpoint::Rendezvous::park_first(
+            &catalog::BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT,
+        );
+        let merge_db = std::sync::Arc::clone(&db);
+        let merge = tokio::spawn(async move { merge_db.branch_merge("s", "t").await });
+        parked.wait_until_reached().await;
+
+        let delete_db = std::sync::Arc::clone(&db);
+        let mut delete = tokio::spawn(async move { delete_db.branch_delete(victim).await });
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_millis(500), &mut delete)
+                .await
+                .is_err(),
+            "deleting {victim} must wait while the merge holds its gate"
+        );
+
+        parked.release();
+        assert_eq!(
+            merge
+                .await
+                .unwrap()
+                .expect("the merge must publish after release"),
+            omnigraph::db::MergeOutcome::Merged,
+            "deleting {victim}"
+        );
+        delete
+            .await
+            .unwrap()
+            .unwrap_or_else(|error| panic!("deleting {victim} after the merge: {error}"));
+        assert!(
+            !db.branch_list().await.unwrap().iter().any(|b| b == victim),
+            "{victim} is gone"
+        );
+    }
+}
+
+/// Issue 643, same-source case: two merges from one source into two targets
+/// both land, and each target gets the source's row. Whether they may overlap
+/// is not pinned: today both take the source's branch gate, so they
+/// serialize, although a merge only reads its source.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn same_source_merges_both_land_issue_643() {
+    let _scenario = FailScenario::setup();
+    let dir = tempfile::tempdir().unwrap();
+    let db = helpers::init_and_load(&dir).await;
+    diverged_merge_pair(&db, "s", "t1").await;
+    db.branch_create("t2").await.unwrap();
+    db.mutate(
+        "t2",
+        MUTATION_QUERIES,
+        "insert_person",
+        &mixed_params(&[("$name", "On-t2")], &[("$age", 54)]),
+    )
+    .await
+    .unwrap();
+    let db = std::sync::Arc::new(db);
+
+    let first_db = std::sync::Arc::clone(&db);
+    let second_db = std::sync::Arc::clone(&db);
+    let (first, second) = tokio::join!(
+        tokio::spawn(async move { first_db.branch_merge("s", "t1").await }),
+        tokio::spawn(async move { second_db.branch_merge("s", "t2").await }),
+    );
+    for (target, outcome) in [("t1", first), ("t2", second)] {
+        assert_eq!(
+            outcome.unwrap().expect("merge must land"),
+            omnigraph::db::MergeOutcome::Merged,
+            "{target}"
+        );
+        assert_eq!(
+            helpers::count_rows_branch(&db, target, "node:Person").await,
+            6,
+            "{target} holds the seed rows, its own insert and the source's"
+        );
+    }
+}
+
 /// Branch create takes the schema gate's EXCLUSIVE side: no CAS covers its
 /// namespace inventory, so a sibling create with disjoint branch gates must
 /// wait at the gate, then refuse on the collision the first leaves behind.

From 156e7a6b93baf9a86bc71baf6753b3bc57485045 Mon Sep 17 00:00:00 2001
From: Ragnor Comerford <hello@ragnor.co>
Date: Wed, 30 Sep 2026 18:36:16 +0200
Subject: [PATCH 04/11] test: a failed merge beside an independent merge

Merge A fails after its table effects, before its manifest commit,
while merge B into another target is parked mid-merge. A's target head
and table pin stay where they were, B publishes on release, and the
same handle retries A to completion, so the failure released A's
gates.
---
 crates/omnigraph/tests/failpoints.rs | 88 ++++++++++++++++++++++++++++
 1 file changed, 88 insertions(+)

diff --git a/crates/omnigraph/tests/failpoints.rs b/crates/omnigraph/tests/failpoints.rs
index 90cbe91db..59ec853e8 100644
--- a/crates/omnigraph/tests/failpoints.rs
+++ b/crates/omnigraph/tests/failpoints.rs
@@ -2175,6 +2175,94 @@ async fn same_source_merges_both_land_issue_643() {
     }
 }
 
+/// Issue 643, failure boundary: merge A fails after its table effects,
+/// before its manifest commit, while merge B into an independent target is
+/// parked mid-merge holding its gates and the shared schema permit. A's
+/// target does not move, B still publishes, and A's gates are released: the
+/// same handle retries A to completion.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn failed_merge_beside_an_independent_merge_issue_643() {
+    let _scenario = FailScenario::setup();
+    let dir = tempfile::tempdir().unwrap();
+    let db = helpers::init_and_load(&dir).await;
+    diverged_merge_pair(&db, "s1", "t1").await;
+    diverged_merge_pair(&db, "s2", "t2").await;
+    let db = std::sync::Arc::new(db);
+    let t1_head = branch_head_commit_id(dir.path(), "t1").await.unwrap();
+    let t1_person = helpers::snapshot_branch(&db, "t1")
+        .await
+        .unwrap()
+        .dataset("node:Person")
+        .unwrap()
+        .published_dataset_version;
+
+    let parked =
+        helpers::failpoint::Rendezvous::park_first(&catalog::BRANCH_MERGE_POST_AUTHORITY_CAPTURE);
+    let merge_b_db = std::sync::Arc::clone(&db);
+    let merge_b = tokio::spawn(async move { merge_b_db.branch_merge("s2", "t2").await });
+    parked.wait_until_reached().await;
+
+    {
+        let _failpoint = catalog::BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT.fire_always();
+        let error = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            db.branch_merge("s1", "t1"),
+        )
+        .await
+        .expect("A must not wait on B's gates")
+        .expect_err("A must fail at its injected fault");
+        assert!(
+            error.to_string().contains(
+                "injected failpoint triggered: branch_merge.post_phase_b_pre_manifest_commit"
+            ),
+            "A must fail at the injected fault, not earlier: {error}"
+        );
+    }
+    assert_eq!(
+        branch_head_commit_id(dir.path(), "t1").await.unwrap(),
+        t1_head,
+        "the failed merge must not move its target's graph head"
+    );
+    assert_eq!(
+        helpers::snapshot_branch(&db, "t1")
+            .await
+            .unwrap()
+            .dataset("node:Person")
+            .unwrap()
+            .published_dataset_version,
+        t1_person,
+        "the failed merge's table effects must stay unpublished"
+    );
+    assert!(
+        !merge_b.is_finished(),
+        "B must still be parked after A failed"
+    );
+
+    parked.release();
+    assert_eq!(
+        merge_b.await.unwrap().expect("B must merge after release"),
+        omnigraph::db::MergeOutcome::Merged
+    );
+    assert_eq!(
+        tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            db.branch_merge("s1", "t1"),
+        )
+        .await
+        .expect("the failed merge must have released its gates")
+        .expect("the same handle must retry A once the fault stops"),
+        omnigraph::db::MergeOutcome::Merged
+    );
+    for target in ["t1", "t2"] {
+        assert_eq!(
+            helpers::count_rows_branch(&db, target, "node:Person").await,
+            6,
+            "{target} holds the seed rows, its own insert and its source's"
+        );
+    }
+}
+
 /// Branch create takes the schema gate's EXCLUSIVE side: no CAS covers its
 /// namespace inventory, so a sibling create with disjoint branch gates must
 /// wait at the gate, then refuse on the collision the first leaves behind.

From dd453d7062f7cddf0e47ace749d9ca1dafd77236 Mon Sep 17 00:00:00 2001
From: Ragnor Comerford <hello@ragnor.co>
Date: Wed, 30 Sep 2026 18:39:46 +0200
Subject: [PATCH 05/11] test: a merge refuses a target recreated by another
 process

Another process deletes a merge's target and creates a branch of the
same name while the merge is parked after its table effects. The merge
refuses with ReadSetChanged on the target's branch identifier, not in
doubt, the recreated target's head does not move, and it holds only
its own writer's rows.
---
 crates/omnigraph/tests/failpoints.rs | 72 ++++++++++++++++++++++++++++
 1 file changed, 72 insertions(+)

diff --git a/crates/omnigraph/tests/failpoints.rs b/crates/omnigraph/tests/failpoints.rs
index 59ec853e8..3a0e0958e 100644
--- a/crates/omnigraph/tests/failpoints.rs
+++ b/crates/omnigraph/tests/failpoints.rs
@@ -229,6 +229,11 @@ fn rfc023_external_writer_process() {
                     .await
                     .unwrap();
                 }
+                "merge_target_recreated" => {
+                    db.branch_delete("t").await.unwrap();
+                    db.branch_create("t").await.unwrap();
+                    db.load("t", &payload, mode).await.unwrap();
+                }
                 other => panic!("unknown external writer action '{other}'"),
             }
         });
@@ -2263,6 +2268,73 @@ async fn failed_merge_beside_an_independent_merge_issue_643() {
     }
 }
 
+/// Issue 643, branch recreation: another process deletes a merge's target
+/// and creates a branch of the same name while the merge is parked after its
+/// table effects, before its manifest commit. The name now denotes a new
+/// branch lifetime, so the merge must refuse rather than publish into it,
+/// and the recreated target keeps exactly what its own writer put there.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn merge_refuses_a_target_recreated_by_another_process_issue_643() {
+    let _scenario = FailScenario::setup();
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().to_str().unwrap().to_string();
+    let db = helpers::init_and_load(&dir).await;
+    diverged_merge_pair(&db, "s", "t").await;
+    let db = std::sync::Arc::new(db);
+
+    let parked = helpers::failpoint::Rendezvous::park_first(
+        &catalog::BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT,
+    );
+    let merge_db = std::sync::Arc::clone(&db);
+    let merge = tokio::spawn(async move { merge_db.branch_merge("s", "t").await });
+    parked.wait_until_reached().await;
+
+    let child_uri = uri.clone();
+    tokio::task::spawn_blocking(move || {
+        run_rfc023_external_writer_action(
+            child_uri,
+            LoadMode::Append,
+            r#"{"type":"Person","data":{"name":"On-t-recreated","age":55}}"#.to_string(),
+            "merge_target_recreated",
+        )
+    })
+    .await
+    .unwrap()
+    .expect("the other process recreates the target");
+    let recreated_head = branch_head_commit_id(dir.path(), "t").await.unwrap();
+
+    parked.release();
+    let error = merge
+        .await
+        .unwrap()
+        .expect_err("the merge must refuse a target recreated under its name");
+    assert!(
+        matches!(
+            &error,
+            OmniError::Manifest(manifest)
+                if !manifest.publication_in_doubt
+                    && matches!(
+                        &manifest.details,
+                        Some(omnigraph::error::ManifestConflictDetails::ReadSetChanged { member, .. })
+                            if member == "branch_identifier:t"
+                    )
+        ),
+        "the refusal must name the target's changed branch identity: {error:?}"
+    );
+    assert_eq!(
+        branch_head_commit_id(dir.path(), "t").await.unwrap(),
+        recreated_head,
+        "the refused merge must not move the recreated target"
+    );
+    let reader = helpers::session(Omnigraph::open(&uri).await.unwrap());
+    assert_eq!(
+        helpers::count_rows_branch(&reader, "t", "node:Person").await,
+        5,
+        "the recreated target holds the seed rows and its own writer's row only"
+    );
+}
+
 /// Branch create takes the schema gate's EXCLUSIVE side: no CAS covers its
 /// namespace inventory, so a sibling create with disjoint branch gates must
 /// wait at the gate, then refuse on the collision the first leaves behind.

From d7b8917916962d0ec464c1706f3abcf7a0da8b98 Mon Sep 17 00:00:00 2001
From: Ragnor Comerford <hello@ragnor.co>
Date: Wed, 30 Sep 2026 18:42:39 +0200
Subject: [PATCH 06/11] test: a cancelled merge leaves its target whole

A merge task aborted while parked at the authority-capture seam or
after its table effects stops at its first await after the release.
The target is untouched or merged whole, the same handle retries the
merge and writes to the target, and cleanup succeeds afterwards.
---
 crates/omnigraph/tests/failpoints.rs | 90 ++++++++++++++++++++++++++++
 1 file changed, 90 insertions(+)

diff --git a/crates/omnigraph/tests/failpoints.rs b/crates/omnigraph/tests/failpoints.rs
index 3a0e0958e..b8b9971f2 100644
--- a/crates/omnigraph/tests/failpoints.rs
+++ b/crates/omnigraph/tests/failpoints.rs
@@ -2268,6 +2268,96 @@ async fn failed_merge_beside_an_independent_merge_issue_643() {
     }
 }
 
+/// Issue 643, cancellation: a merge whose task is aborted while it is parked
+/// at a seam stops at its first await after the release. On a local store
+/// that falls before its manifest commit, after its table effects at the
+/// later seam; the assertions also accept a merge that published before the
+/// cancel took effect. Either way the target is whole (untouched, or merged
+/// with a new head), the handle stays live (the retry completes and another
+/// write lands), and cleanup succeeds afterwards.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn cancelled_merge_leaves_the_target_whole_issue_643() {
+    for seam in [
+        &catalog::BRANCH_MERGE_POST_AUTHORITY_CAPTURE,
+        &catalog::BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT,
+    ] {
+        let _scenario = FailScenario::setup();
+        let dir = tempfile::tempdir().unwrap();
+        let db = helpers::init_and_load(&dir).await;
+        diverged_merge_pair(&db, "s", "t").await;
+        let db = std::sync::Arc::new(db);
+        let head_before = branch_head_commit_id(dir.path(), "t").await.unwrap();
+
+        let parked = helpers::failpoint::Rendezvous::park_first(seam);
+        let merge_db = std::sync::Arc::clone(&db);
+        let merge = tokio::spawn(async move { merge_db.branch_merge("s", "t").await });
+        parked.wait_until_reached().await;
+        merge.abort();
+        parked.release();
+        let outcome = merge.await;
+        drop(parked);
+
+        let merged = branch_head_commit_id(dir.path(), "t").await.unwrap() != head_before;
+        let rows = helpers::count_rows_branch(&db, "t", "node:Person").await;
+        let seam = seam.name();
+        match &outcome {
+            Ok(result) => assert_eq!(
+                result.as_ref().expect("an uncancelled merge must land"),
+                &omnigraph::db::MergeOutcome::Merged,
+                "{seam}"
+            ),
+            Err(join) => assert!(join.is_cancelled(), "{seam}: {join}"),
+        }
+        if outcome.is_ok() {
+            assert!(
+                merged,
+                "{seam}: an acknowledged merge moves the target head"
+            );
+        }
+        assert_eq!(
+            rows,
+            if merged { 6 } else { 5 },
+            "{seam}: the target is merged whole or untouched"
+        );
+
+        let retried = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            db.branch_merge("s", "t"),
+        )
+        .await
+        .unwrap_or_else(|_| panic!("{seam}: the cancelled merge must have released its gates"))
+        .unwrap_or_else(|error| panic!("{seam}: the retry must merge: {error}"));
+        assert_eq!(
+            retried,
+            if merged {
+                omnigraph::db::MergeOutcome::AlreadyUpToDate
+            } else {
+                omnigraph::db::MergeOutcome::Merged
+            },
+            "{seam}"
+        );
+        db.mutate(
+            "t",
+            MUTATION_QUERIES,
+            "insert_person",
+            &mixed_params(&[("$name", "After-cancel")], &[("$age", 56)]),
+        )
+        .await
+        .unwrap_or_else(|error| panic!("{seam}: a later write on the target: {error}"));
+        let rows = db.cleanup(keep_one()).await.unwrap();
+        assert!(
+            rows.iter().all(|row| row.error.is_none()),
+            "{seam}: {rows:?}"
+        );
+        assert_eq!(
+            helpers::count_rows_branch(&db, "t", "node:Person").await,
+            7,
+            "{seam}: seed rows, both inserts, the merged row and the later write"
+        );
+    }
+}
+
 /// Issue 643, branch recreation: another process deletes a merge's target
 /// and creates a branch of the same name while the merge is parked after its
 /// table effects, before its manifest commit. The name now denotes a new

From dc37cd775ca426301d9b7000d62fbac1d0beaf63 Mon Sep 17 00:00:00 2001
From: Ragnor Comerford <hello@ragnor.co>
Date: Thu, 1 Oct 2026 13:54:45 +0200
Subject: [PATCH 07/11] test: compare merge outcomes through the publication
 receipt

---
 crates/omnigraph/tests/failpoints.rs   | 41 +++++++++++++++++++-------
 crates/omnigraph/tests/schema_apply.rs |  3 +-
 2 files changed, 33 insertions(+), 11 deletions(-)

diff --git a/crates/omnigraph/tests/failpoints.rs b/crates/omnigraph/tests/failpoints.rs
index ba732e6e6..04c4abd07 100644
--- a/crates/omnigraph/tests/failpoints.rs
+++ b/crates/omnigraph/tests/failpoints.rs
@@ -2015,7 +2015,7 @@ async fn independent_target_merges_overlap_issue_643() {
         .expect("a merge into another target must complete while A is parked")
         .unwrap()
         .expect("B must merge");
-    assert_eq!(merged_b, omnigraph::db::MergeOutcome::Merged);
+    assert_eq!(merged_b.outcome, omnigraph::db::MergeOutcome::Merged);
     assert!(
         !merge_a.is_finished(),
         "A must still be parked before its manifest commit while B published"
@@ -2023,7 +2023,11 @@ async fn independent_target_merges_overlap_issue_643() {
 
     parked.release();
     assert_eq!(
-        merge_a.await.unwrap().expect("A must merge after release"),
+        merge_a
+            .await
+            .unwrap()
+            .expect("A must merge after release")
+            .outcome,
         omnigraph::db::MergeOutcome::Merged
     );
     for target in ["t1", "t2"] {
@@ -2075,11 +2079,19 @@ async fn same_target_merges_serialize_issue_643() {
 
     parked.release();
     assert_eq!(
-        merge_a.await.unwrap().expect("A must merge after release"),
+        merge_a
+            .await
+            .unwrap()
+            .expect("A must merge after release")
+            .outcome,
         omnigraph::db::MergeOutcome::Merged
     );
     assert_eq!(
-        merge_b.await.unwrap().expect("B must merge after A"),
+        merge_b
+            .await
+            .unwrap()
+            .expect("B must merge after A")
+            .outcome,
         omnigraph::db::MergeOutcome::Merged
     );
     assert_eq!(
@@ -2123,7 +2135,8 @@ async fn merge_endpoint_delete_waits_for_the_merge_issue_643() {
             merge
                 .await
                 .unwrap()
-                .expect("the merge must publish after release"),
+                .expect("the merge must publish after release")
+                .outcome,
             omnigraph::db::MergeOutcome::Merged,
             "deleting {victim}"
         );
@@ -2168,7 +2181,7 @@ async fn same_source_merges_both_land_issue_643() {
     );
     for (target, outcome) in [("t1", first), ("t2", second)] {
         assert_eq!(
-            outcome.unwrap().expect("merge must land"),
+            outcome.unwrap().expect("merge must land").outcome,
             omnigraph::db::MergeOutcome::Merged,
             "{target}"
         );
@@ -2246,7 +2259,11 @@ async fn failed_merge_beside_an_independent_merge_issue_643() {
 
     parked.release();
     assert_eq!(
-        merge_b.await.unwrap().expect("B must merge after release"),
+        merge_b
+            .await
+            .unwrap()
+            .expect("B must merge after release")
+            .outcome,
         omnigraph::db::MergeOutcome::Merged
     );
     assert_eq!(
@@ -2256,7 +2273,8 @@ async fn failed_merge_beside_an_independent_merge_issue_643() {
         )
         .await
         .expect("the failed merge must have released its gates")
-        .expect("the same handle must retry A once the fault stops"),
+        .expect("the same handle must retry A once the fault stops")
+        .outcome,
         omnigraph::db::MergeOutcome::Merged
     );
     for target in ["t1", "t2"] {
@@ -2303,7 +2321,10 @@ async fn cancelled_merge_leaves_the_target_whole_issue_643() {
         let seam = seam.name();
         match &outcome {
             Ok(result) => assert_eq!(
-                result.as_ref().expect("an uncancelled merge must land"),
+                &result
+                    .as_ref()
+                    .expect("an uncancelled merge must land")
+                    .outcome,
                 &omnigraph::db::MergeOutcome::Merged,
                 "{seam}"
             ),
@@ -2329,7 +2350,7 @@ async fn cancelled_merge_leaves_the_target_whole_issue_643() {
         .unwrap_or_else(|_| panic!("{seam}: the cancelled merge must have released its gates"))
         .unwrap_or_else(|error| panic!("{seam}: the retry must merge: {error}"));
         assert_eq!(
-            retried,
+            retried.outcome,
             if merged {
                 omnigraph::db::MergeOutcome::AlreadyUpToDate
             } else {
diff --git a/crates/omnigraph/tests/schema_apply.rs b/crates/omnigraph/tests/schema_apply.rs
index bdbd2ce71..059ca95c0 100644
--- a/crates/omnigraph/tests/schema_apply.rs
+++ b/crates/omnigraph/tests/schema_apply.rs
@@ -540,7 +540,8 @@ async fn parked_merge_blocks_schema_apply_issue_643() {
         merge
             .await
             .unwrap()
-            .expect("the merge must publish after release"),
+            .expect("the merge must publish after release")
+            .outcome,
         omnigraph::db::MergeOutcome::Merged
     );
     let refusal = schema_task

From f1905989166ce6af55668727ad302a4f920d4266 Mon Sep 17 00:00:00 2001
From: Ragnor Comerford <hello@ragnor.co>
Date: Thu, 1 Oct 2026 17:00:04 +0200
Subject: [PATCH 08/11] merge: stop holding the source branch through the merge

A merge held its source's branch gate and every per-table queue on the
source from preparation through publication, so writes, index builds and
deletes on the source waited for the whole merge, and merges from one
source into several targets ran one after another. Its input is the
tagged source snapshot: the merge pins it, proves it was still current
after tagging, and the final re-read still refuses a source whose
incarnation or schema changed.

Hold the source gate only through that capture, and take per-table
queues for the target only. The target gate and queues stay through
publication, keeping a target delete/recreate out from under the plan
and serializing merges into one target.
---
 ...leases-its-source-after-capture.changed.md |   6 +
 crates/omnigraph/src/exec/merge.rs            |  44 +++--
 crates/omnigraph/tests/failpoints.rs          | 170 +++++++++++++++---
 docs/dev/writes.md                            |   5 +-
 4 files changed, 184 insertions(+), 41 deletions(-)
 create mode 100644 changelog.d/merge-releases-its-source-after-capture.changed.md

diff --git a/changelog.d/merge-releases-its-source-after-capture.changed.md b/changelog.d/merge-releases-its-source-after-capture.changed.md
new file mode 100644
index 000000000..cdf943b7e
--- /dev/null
+++ b/changelog.d/merge-releases-its-source-after-capture.changed.md
@@ -0,0 +1,6 @@
+- Writes and deletes on a merge's source branch no longer wait for the merge.
+  A merge holds its source branch only while it captures the source, then
+  publishes that captured state; changes made to the source afterwards reach
+  the target with the next merge. Merges from one source into several
+  targets now run at the same time. Merges into one target still run one
+  after another.
diff --git a/crates/omnigraph/src/exec/merge.rs b/crates/omnigraph/src/exec/merge.rs
index f3f15a6db..b9e54e5cb 100644
--- a/crates/omnigraph/src/exec/merge.rs
+++ b/crates/omnigraph/src/exec/merge.rs
@@ -5299,10 +5299,28 @@ impl Omnigraph {
 
         let relevant_branches = [source_branch.as_deref(), target_branch.as_deref()];
         let _schema_permit = self.write_queue().acquire_schema_shared().await;
-        let _branch_guards = self
-            .write_queue()
-            .acquire_branches(&[source_branch.clone(), target_branch.clone()])
-            .await;
+        // The target gate is held through publication: it keeps a target
+        // delete/recreate from reusing the name underneath the plan (ABA) and
+        // serializes merges into one target. The source gate is held only
+        // until the source snapshot is captured, tagged and proved still
+        // current below; from then on the tagged snapshot is the merge's
+        // input, so writes and deletes on the source do not wait for the merge.
+        // Both are taken in the sorted order `acquire_branches` uses, so the
+        // pair cannot deadlock with another multi-branch acquirer.
+        let queue = self.write_queue();
+        let (source_gate, _target_gate) = if source_branch <= target_branch {
+            let source_gate = queue.acquire_branch(source_branch.as_deref()).await;
+            (
+                source_gate,
+                queue.acquire_branch(target_branch.as_deref()).await,
+            )
+        } else {
+            let target_gate = queue.acquire_branch(target_branch.as_deref()).await;
+            (
+                queue.acquire_branch(source_branch.as_deref()).await,
+                target_gate,
+            )
+        };
         // Capture each branch as one coherent RFC-022 authority token plus
         // immutable snapshot. The target token is the coarse publish read set;
         // the source token pins the exact merge input without requiring the
@@ -5378,6 +5396,7 @@ impl Omnigraph {
                 return Err(error);
             }
         };
+        drop(source_gate);
         // The handle remains bound to its original branch throughout the merge.
         // The captured transaction supplies every physical and publish target.
         let target_was_active = self.active_branch().await == target_branch;
@@ -5781,18 +5800,17 @@ impl Omnigraph {
         //
         // This bridge still coexists with legacy maintenance writers that take
         // only `(table, branch)` queues. Acquire the conservative all-catalog
-        // envelope for BOTH source and target, in the global sorted order, then
-        // re-read both manifest branches before the first effect. Planning
-        // remains outside table queues, but no plan derived from a stale
-        // source/target snapshot can cross into physical effects.
+        // envelope for the target, in the global sorted order, then re-read
+        // both manifest branches before the first effect. Planning remains
+        // outside table queues, but no plan derived from a stale target
+        // snapshot can cross into physical effects. The source's queues are
+        // not taken: its input is the tagged snapshot captured above, so a
+        // write, index build or delete on the source may proceed; the re-read
+        // still refuses a source whose incarnation or schema changed by then.
         let final_revalidation_timing = crate::instrumentation::start_merge_timing(
             crate::instrumentation::MergeTimingPhase::FinalRevalidation,
         );
-        let active_branch_for_keys = target_branch.map(str::to_string);
-        let merge_branches = [
-            source_branch.map(str::to_string),
-            active_branch_for_keys.clone(),
-        ];
+        let merge_branches = [target_branch.map(str::to_string)];
         let merge_queue_keys = self.table_queue_keys_for_branches(&merge_branches, catalog);
         let _merge_queue_guards = self.write_queue().acquire_many(&merge_queue_keys).await;
 
diff --git a/crates/omnigraph/tests/failpoints.rs b/crates/omnigraph/tests/failpoints.rs
index e11fb00c1..4b1fc904b 100644
--- a/crates/omnigraph/tests/failpoints.rs
+++ b/crates/omnigraph/tests/failpoints.rs
@@ -2139,17 +2139,31 @@ async fn same_target_merges_serialize_issue_643() {
     );
 }
 
-/// Issue 643, delete boundary: deleting a merge's target or source while
-/// the merge is parked before its manifest commit waits on that branch's
-/// gate. The merge then publishes, and the delete succeeds after it.
+/// Issue 643, delete boundary. Deleting a merge's target while the merge is
+/// parked before its manifest commit waits on the target's gate; the merge
+/// publishes, then the delete succeeds. Deleting its source does not wait:
+/// the merge released the source's gate once its tagged source snapshot was
+/// captured. The merge still publishes that snapshot, and the target keeps
+/// reading the merged row after cleanup reclaims what the deletion retired.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 #[serial]
-async fn merge_endpoint_delete_waits_for_the_merge_issue_643() {
+async fn merge_target_delete_waits_and_source_delete_does_not_issue_643() {
     for victim in ["t", "s"] {
         let _scenario = FailScenario::setup();
         let dir = tempfile::tempdir().unwrap();
         let db = helpers::init_and_load(&dir).await;
         diverged_merge_pair(&db, "s", "t").await;
+        // An edge only the source adds: the merge adopts the source's
+        // `edge:Knows` pin as the target's, the case a reclaimed source would
+        // break.
+        db.mutate(
+            "s",
+            MUTATION_QUERIES,
+            "add_friend",
+            &params(&[("$from", "Diana"), ("$to", "Alice")]),
+        )
+        .await
+        .unwrap();
         let db = std::sync::Arc::new(db);
 
         let parked = helpers::failpoint::Rendezvous::park_first(
@@ -2161,12 +2175,18 @@ async fn merge_endpoint_delete_waits_for_the_merge_issue_643() {
 
         let delete_db = std::sync::Arc::clone(&db);
         let mut delete = tokio::spawn(async move { delete_db.branch_delete(victim).await });
-        assert!(
-            tokio::time::timeout(std::time::Duration::from_millis(500), &mut delete)
-                .await
-                .is_err(),
-            "deleting {victim} must wait while the merge holds its gate"
-        );
+        let early = tokio::time::timeout(std::time::Duration::from_millis(500), &mut delete).await;
+        if victim == "t" {
+            assert!(
+                early.is_err(),
+                "deleting the target must wait while the merge holds its gate"
+            );
+        } else {
+            early
+                .expect("deleting the source must not wait for the merge")
+                .unwrap()
+                .unwrap_or_else(|error| panic!("deleting the source during the merge: {error}"));
+        }
 
         parked.release();
         assert_eq!(
@@ -2178,10 +2198,25 @@ async fn merge_endpoint_delete_waits_for_the_merge_issue_643() {
             omnigraph::db::MergeOutcome::Merged,
             "deleting {victim}"
         );
-        delete
-            .await
-            .unwrap()
-            .unwrap_or_else(|error| panic!("deleting {victim} after the merge: {error}"));
+        if victim == "t" {
+            delete
+                .await
+                .unwrap()
+                .unwrap_or_else(|error| panic!("deleting the target after the merge: {error}"));
+        } else {
+            let rows = db.cleanup(keep_one()).await.unwrap();
+            assert!(rows.iter().all(|row| row.error.is_none()), "{rows:?}");
+            assert_eq!(
+                helpers::count_rows_branch(&db, "t", "node:Person").await,
+                6,
+                "t holds the seed rows, its own insert and the deleted source's"
+            );
+            assert_eq!(
+                helpers::count_rows_branch(&db, "t", "edge:Knows").await,
+                4,
+                "t reads the edge adopted from the deleted source after cleanup"
+            );
+        }
         assert!(
             !db.branch_list().await.unwrap().iter().any(|b| b == victim),
             "{victim} is gone"
@@ -2189,13 +2224,13 @@ async fn merge_endpoint_delete_waits_for_the_merge_issue_643() {
     }
 }
 
-/// Issue 643, same-source case: two merges from one source into two targets
-/// both land, and each target gets the source's row. Whether they may overlap
-/// is not pinned: today both take the source's branch gate, so they
-/// serialize, although a merge only reads its source.
+/// Issue 643, same-source case: with a merge from `s` into `t1` parked
+/// before its manifest commit, a merge from `s` into `t2` runs to
+/// completion, because a merge holds its source's gate only through
+/// capture. Both land, and each target gets the source's row.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 #[serial]
-async fn same_source_merges_both_land_issue_643() {
+async fn same_source_merges_overlap_issue_643() {
     let _scenario = FailScenario::setup();
     let dir = tempfile::tempdir().unwrap();
     let db = helpers::init_and_load(&dir).await;
@@ -2211,18 +2246,35 @@ async fn same_source_merges_both_land_issue_643() {
     .unwrap();
     let db = std::sync::Arc::new(db);
 
+    let parked = helpers::failpoint::Rendezvous::park_first(
+        &catalog::BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT,
+    );
     let first_db = std::sync::Arc::clone(&db);
+    let first = tokio::spawn(async move { first_db.branch_merge("s", "t1").await });
+    parked.wait_until_reached().await;
+
     let second_db = std::sync::Arc::clone(&db);
-    let (first, second) = tokio::join!(
-        tokio::spawn(async move { first_db.branch_merge("s", "t1").await }),
+    let second = tokio::time::timeout(
+        std::time::Duration::from_secs(10),
         tokio::spawn(async move { second_db.branch_merge("s", "t2").await }),
+    )
+    .await
+    .expect("a second merge from the same source must complete while the first is parked")
+    .unwrap()
+    .expect("the second merge must land");
+    assert_eq!(second.outcome, omnigraph::db::MergeOutcome::Merged);
+    assert!(!first.is_finished(), "the first merge is still parked");
+
+    parked.release();
+    assert_eq!(
+        first
+            .await
+            .unwrap()
+            .expect("the first merge must land")
+            .outcome,
+        omnigraph::db::MergeOutcome::Merged
     );
-    for (target, outcome) in [("t1", first), ("t2", second)] {
-        assert_eq!(
-            outcome.unwrap().expect("merge must land").outcome,
-            omnigraph::db::MergeOutcome::Merged,
-            "{target}"
-        );
+    for target in ["t1", "t2"] {
         assert_eq!(
             helpers::count_rows_branch(&db, target, "node:Person").await,
             6,
@@ -2231,6 +2283,70 @@ async fn same_source_merges_both_land_issue_643() {
     }
 }
 
+/// Issue 643, writes on a merge's source: with a merge from `s` into `t`
+/// parked before its manifest commit, an insert on `s` completes. The merge
+/// publishes the source snapshot it captured, so the concurrent insert is not
+/// in `t` until a second merge carries it.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn source_write_proceeds_during_merge_issue_643() {
+    let _scenario = FailScenario::setup();
+    let dir = tempfile::tempdir().unwrap();
+    let db = helpers::init_and_load(&dir).await;
+    diverged_merge_pair(&db, "s", "t").await;
+    let db = std::sync::Arc::new(db);
+
+    let parked = helpers::failpoint::Rendezvous::park_first(
+        &catalog::BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT,
+    );
+    let merge_db = std::sync::Arc::clone(&db);
+    let merge = tokio::spawn(async move { merge_db.branch_merge("s", "t").await });
+    parked.wait_until_reached().await;
+
+    tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        db.mutate(
+            "s",
+            MUTATION_QUERIES,
+            "insert_person",
+            &mixed_params(&[("$name", "During-merge")], &[("$age", 63)]),
+        ),
+    )
+    .await
+    .expect("an insert on the merge's source must not wait for the merge")
+    .expect("the insert on the source must land");
+    assert!(!merge.is_finished(), "the merge is still parked");
+
+    parked.release();
+    assert_eq!(
+        merge
+            .await
+            .unwrap()
+            .expect("the merge must publish after release")
+            .outcome,
+        omnigraph::db::MergeOutcome::Merged
+    );
+    assert_eq!(
+        helpers::count_rows_branch(&db, "s", "node:Person").await,
+        6,
+        "s holds the seed rows, its own insert and the concurrent one"
+    );
+    assert_eq!(
+        helpers::count_rows_branch(&db, "t", "node:Person").await,
+        6,
+        "t holds the seed rows, its own insert and the source as captured"
+    );
+    assert_eq!(
+        db.branch_merge("s", "t").await.unwrap().outcome,
+        omnigraph::db::MergeOutcome::Merged
+    );
+    assert_eq!(
+        helpers::count_rows_branch(&db, "t", "node:Person").await,
+        7,
+        "a second merge carries the concurrent insert"
+    );
+}
+
 /// Issue 643, failure boundary: merge A fails after its table effects,
 /// before its manifest commit, while merge B into an independent target is
 /// parked mid-merge holding its gates and the shared schema permit. A's
diff --git a/docs/dev/writes.md b/docs/dev/writes.md
index 4df9475c2..4dfde5172 100644
--- a/docs/dev/writes.md
+++ b/docs/dev/writes.md
@@ -58,7 +58,10 @@ and enforces the exact graph-head precondition on every attempt. Successful
 publication returns a taken coordinator to the one-entry merge cache; failure
 drops it. Commit IDs and timestamps are minted for the captured branch without
 reloading manifest history. The existing schema and branch gates still serialize
-conflicting control operations.
+conflicting control operations. A merge holds its target's branch gate through
+publication and its source's only until the source snapshot is captured, tagged
+and proved still current, so writes and deletes on the source do not wait for
+the merge.
 
 Native branch creation uses an operation-local capture of the bound coordinator
 or that same one-entry cache after the control gates. Reuse

From 0ab3af7cd91b4901282e1c5d4f7f13d3278fe4e6 Mon Sep 17 00:00:00 2001
From: Ragnor Comerford <hello@ragnor.co>
Date: Thu, 1 Oct 2026 17:24:44 +0200
Subject: [PATCH 09/11] docs: the merge's source and target locks

---
 docs/dev/writes.md | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git a/docs/dev/writes.md b/docs/dev/writes.md
index 4dfde5172..441346308 100644
--- a/docs/dev/writes.md
+++ b/docs/dev/writes.md
@@ -58,10 +58,12 @@ and enforces the exact graph-head precondition on every attempt. Successful
 publication returns a taken coordinator to the one-entry merge cache; failure
 drops it. Commit IDs and timestamps are minted for the captured branch without
 reloading manifest history. The existing schema and branch gates still serialize
-conflicting control operations. A merge holds its target's branch gate through
-publication and its source's only until the source snapshot is captured, tagged
-and proved still current, so writes and deletes on the source do not wait for
-the merge.
+conflicting control operations. A merge holds its target's branch gate and
+per-table queues through publication, and its source's branch gate only until
+the source snapshot is captured, tagged and proved still current; it takes no
+per-table queue on the source. Writes, index builds and deletes on the source
+therefore do not wait for the merge, and the merge's final re-read still
+refuses a source whose incarnation or schema changed.
 
 Native branch creation uses an operation-local capture of the bound coordinator
 or that same one-entry cache after the control gates. Reuse

From 2879686f7ce1b632ee29a5d486e0c34d8f155cf5 Mon Sep 17 00:00:00 2001
From: Ragnor Comerford <hello@ragnor.co>
Date: Sun, 11 Oct 2026 22:12:48 -0700
Subject: [PATCH 10/11] merge: prove pure inserts against the captured source
 pin

A merge that releases its source gate after capture let a source write
land during planning, and the pure-insert revalidation then refused the
merge because the live source pin no longer equalled the proven one. The
proof reads the source at the captured version the merge-input tag
retains, so the check keeps identity, path and native ref and drops the
version equality; its message now reports the live pin.

The source-write test gains a second half that parks at candidate
validation on a merge whose Person change is proven insert-only, lands a
source insert, and requires the merge to publish the captured rows.
---
 crates/omnigraph/src/exec/merge.rs   |  6 ++-
 crates/omnigraph/tests/failpoints.rs | 79 +++++++++++++++++++++++++++-
 docs/dev/writes.md                   |  7 ++-
 3 files changed, 87 insertions(+), 5 deletions(-)

diff --git a/crates/omnigraph/src/exec/merge.rs b/crates/omnigraph/src/exec/merge.rs
index 0db774323..004145af4 100644
--- a/crates/omnigraph/src/exec/merge.rs
+++ b/crates/omnigraph/src/exec/merge.rs
@@ -1416,10 +1416,12 @@ async fn revalidate_proven_pure_insert_source(
             None,
         ));
     };
+    // The proof reads `proven.source`, opened at the captured version the
+    // merge-input tag retains, so a later pin on the same table is not a
+    // change of input.
     if live_entry.identity != proven.source_identity
         || live_entry.dataset_path != proven.source_table_path
         || live_entry.native_dataset_branch != proven.source_table_branch
-        || pin_version(live_entry) != proven.source_version
     {
         return Err(OmniError::manifest_read_set_changed(
             format!("branch_merge_source_dataset:{table_key}"),
@@ -1435,7 +1437,7 @@ async fn revalidate_proven_pure_insert_source(
                 live_entry.identity,
                 live_entry.dataset_path,
                 live_entry.native_dataset_branch,
-                live_entry.published_dataset_version
+                pin_version(live_entry)
             )),
         ));
     }
diff --git a/crates/omnigraph/tests/failpoints.rs b/crates/omnigraph/tests/failpoints.rs
index 2643c1b73..fc83dd398 100644
--- a/crates/omnigraph/tests/failpoints.rs
+++ b/crates/omnigraph/tests/failpoints.rs
@@ -2605,7 +2605,10 @@ async fn same_source_merges_overlap_issue_643() {
 /// Issue 643, writes on a merge's source: with a merge from `s` into `t`
 /// parked before its manifest commit, an insert on `s` completes. The merge
 /// publishes the source snapshot it captured, so the concurrent insert is not
-/// in `t` until a second merge carries it.
+/// in `t` until a second merge carries it. The second half parks earlier, at
+/// candidate validation, on a merge whose `Person` change is proven
+/// insert-only: the source table moving past the proven version must not
+/// refuse a merge whose input is the captured, tagged version.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 #[serial]
 async fn source_write_proceeds_during_merge_issue_643() {
@@ -2664,6 +2667,80 @@ async fn source_write_proceeds_during_merge_issue_643() {
         7,
         "a second merge carries the concurrent insert"
     );
+
+    // The target diverges on `edge:Knows` only, so its `Person` equals the
+    // base and the source's `Person` inserts take the proven pure-insert path.
+    let dir = tempfile::tempdir().unwrap();
+    let db = helpers::init_and_load(&dir).await;
+    db.branch_create("s").await.unwrap();
+    db.branch_create("t").await.unwrap();
+    db.mutate(
+        "s",
+        MUTATION_QUERIES,
+        "insert_person",
+        &mixed_params(&[("$name", "On-s")], &[("$age", 51)]),
+    )
+    .await
+    .unwrap();
+    db.mutate(
+        "t",
+        MUTATION_QUERIES,
+        "add_friend",
+        &params(&[("$from", "Diana"), ("$to", "Alice")]),
+    )
+    .await
+    .unwrap();
+    let db = std::sync::Arc::new(db);
+
+    let parked = helpers::failpoint::Rendezvous::park_first(
+        &catalog::BRANCH_MERGE_POST_CANDIDATE_VALIDATION,
+    );
+    let probes = MergeWriteProbes::default();
+    let merge_db = std::sync::Arc::clone(&db);
+    let merge = tokio::spawn(with_merge_write_probes(probes.clone(), async move {
+        merge_db.branch_merge("s", "t").await
+    }));
+    parked.wait_until_reached().await;
+    tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        db.mutate(
+            "s",
+            MUTATION_QUERIES,
+            "insert_person",
+            &mixed_params(&[("$name", "During-validation")], &[("$age", 64)]),
+        ),
+    )
+    .await
+    .expect("an insert on the merge's source must not wait for candidate validation")
+    .expect("the insert on the source must land");
+
+    parked.release();
+    assert_eq!(
+        merge
+            .await
+            .unwrap()
+            .expect("the source moving past the proven version must not refuse the merge")
+            .outcome,
+        omnigraph::db::MergeOutcome::Merged
+    );
+    assert!(
+        probes.proven_insert_history_read_calls() >= 1,
+        "the merge must have certified the source's Person insertions"
+    );
+    assert_eq!(
+        helpers::count_rows_branch(&db, "t", "node:Person").await,
+        5,
+        "t holds the seed rows and the source's Person as captured"
+    );
+    assert_eq!(
+        db.branch_merge("s", "t").await.unwrap().outcome,
+        omnigraph::db::MergeOutcome::Merged
+    );
+    assert_eq!(
+        helpers::count_rows_branch(&db, "t", "node:Person").await,
+        6,
+        "a second merge carries the insert made during validation"
+    );
 }
 
 /// Issue 643, failure boundary: merge A fails after its table effects,
diff --git a/docs/dev/writes.md b/docs/dev/writes.md
index 120425bc3..758e95ed5 100644
--- a/docs/dev/writes.md
+++ b/docs/dev/writes.md
@@ -362,8 +362,11 @@ table's pin naming `base + 1` as its `published_dataset_version` and the
 chain's tip as its `staged_version`, whatever the chain's length. The
 pure-insert proof walks the branch's commit chain by `read_version` links
 from the source pin back to the base pin (`try_proven_pure_insert_history`,
-`proven_chain_fragments` in `exec/merge.rs`); lineage candidate discovery
-(`plan_lineage_merge`) compares the base and side manifests and walks no
+`proven_chain_fragments` in `exec/merge.rs`). The source pin is the
+captured one the merge-input tag retains: a write to the source table after
+capture does not refuse the merge, while a source table whose identity, path
+or native ref changed does (`revalidate_proven_pure_insert_source`).
+Lineage candidate discovery (`plan_lineage_merge`) compares the base and side manifests and walks no
 chain. A failure anywhere before publication leaves the target untouched
 and the chain as unpublished staging; a target that advanced meanwhile
 makes the merge lose its manifest CAS and return the ordinary conflict.

From 3a1c7a50457c2b443d3f91002ac121eadcca9dda Mon Sep 17 00:00:00 2001
From: Ragnor Comerford <hello@ragnor.co>
Date: Sun, 11 Oct 2026 22:12:54 -0700
Subject: [PATCH 11/11] docs: a merge waiting for its target holds a source
 that sorts first

Gates are taken in sorted branch order, so a merge whose source sorts
first (main always does) holds the source gate while another merge holds
its target. The guide and the release note no longer say source writes
never wait for a merge; they wait only until the merge captures its source.
---
 ...rge-releases-its-source-after-capture.changed.md | 13 +++++++------
 docs/dev/writes.md                                  |  9 ++++++---
 2 files changed, 13 insertions(+), 9 deletions(-)

diff --git a/changelog.d/merge-releases-its-source-after-capture.changed.md b/changelog.d/merge-releases-its-source-after-capture.changed.md
index cdf943b7e..077830b8e 100644
--- a/changelog.d/merge-releases-its-source-after-capture.changed.md
+++ b/changelog.d/merge-releases-its-source-after-capture.changed.md
@@ -1,6 +1,7 @@
-- Writes and deletes on a merge's source branch no longer wait for the merge.
-  A merge holds its source branch only while it captures the source, then
-  publishes that captured state; changes made to the source afterwards reach
-  the target with the next merge. Merges from one source into several
-  targets now run at the same time. Merges into one target still run one
-  after another.
+- Writes and deletes on a merge's source branch no longer wait for the whole
+  merge. A merge holds its source branch only while it captures the source,
+  then publishes that captured state; changes made to the source afterwards
+  reach the target with the next merge. A merge still waiting for its target,
+  because another merge into that target is running, can hold its source
+  until it starts. Merges from one source into several targets now run at
+  the same time. Merges into one target still run one after another.
diff --git a/docs/dev/writes.md b/docs/dev/writes.md
index 758e95ed5..53aabcad9 100644
--- a/docs/dev/writes.md
+++ b/docs/dev/writes.md
@@ -62,9 +62,12 @@ addressable commit id `hb1.<block>.<slot>.<nonce>`. The existing schema and bran
 conflicting control operations. A merge holds its target's branch gate and
 per-table queues through publication, and its source's branch gate only until
 the source snapshot is captured, tagged and proved still current; it takes no
-per-table queue on the source. Writes, index builds and deletes on the source
-therefore do not wait for the merge, and the merge's final re-read still
-refuses a source whose incarnation or schema changed.
+per-table queue on the source. Both gates are taken in sorted branch order, so
+a merge whose source sorts first (`main` always does) holds the source gate
+while it waits for a target another merge holds. Once the source is captured,
+writes, index builds and deletes on the source do not wait for the merge, and
+the merge's final re-read still refuses a source whose incarnation or schema
+changed.
 
 Native branch creation uses an operation-local capture of the bound coordinator
 or that same one-entry cache after the control gates. Reuse