jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-06 17:31:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(buzz-db): replace barrier race with controlled lock-wait-observation schedule

Browse Source 
The previous concurrency regression used a tokio::sync::Barrier to
synchronize two connections before racing to INSERT the same revision.
That construction synchronizes client-side INSERT dispatch, not trigger
execution; a lock-free schedule where one INSERT completes and commits
before the other reads MAX still satisfies the XOR + 23514 assertions,
so the test offered no deterministic proof that pg_advisory_xact_lock
is required.

Replace with a controlled two-connection schedule:

  1. tx1 opens a transaction and inserts revision 102. The BEFORE INSERT
     trigger acquires pg_advisory_xact_lock and completes; tx1 holds the
     advisory lock until commit.
  2. tx2 opens a transaction on a second backend, reports its pg_backend_pid
     over a oneshot channel, then issues INSERT for revision 103. The
     trigger fires and blocks on the advisory lock held by tx1.
  3. The main task polls pg_stat_activity WHERE pid = tx2_pid AND
     wait_event_type = 'Lock' AND wait_event = 'advisory' with a 10 s
     bounded timeout. Without pg_advisory_xact_lock in the guard the
     trigger returns immediately, tx2 never enters the advisory wait, and
     the poll times out — making the regression deterministically red.
  4. tx1 commits, releasing the lock. tx2 unblocks, its trigger reads the
     fresh MAX=102, and INSERT 103 succeeds. tx2 commits.
  5. Final count asserts six revisions (1, 2, 100, 101, 102, 103).

Zero production changes: migrations/0041, migrations/0042, and
schema/schema.sql are byte-untouched (single-file diff confirmed).

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Hayt <9e1c23a3fd83f61da34420e4e88ff1b16e45cafcc0cd9019eb07d4ecfa8ca9b0@buzz.block.builderlab.xyz>
This commit is contained in:
Hayt
2026-08-29 19:15:47 -04:00
parent c52444b544
commit 0534277b4a
1 changed files with 110 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

203
crates/buzz-db/src/runtime/migration.rs
View File

@@ -3345,113 +3345,130 @@ mod tests {
got: {replay_err}" got: {replay_err}"
); );
// Concurrency regression: prove the advisory lock actually serializes // Concurrency regression: prove the advisory lock is load-bearing. The
// concurrent writers. Two connections race to insert the SAME next revision // test uses a controlled two-connection schedule:
// (102) for the same community. The advisory lock must cause one writer to
// block, see the other's committed MAX, and then be rejected with 23514 from
// the named guard. Without the lock, both could pass the MAX check before
// either commits; only a PK collision (23505) would catch the duplicate —
// not the guard. Removing pg_advisory_xact_lock from the guard function and
// re-running must produce 23505 (PK) rather than 23514 (guard), proving the
// test is mutation-sensitive to the lock.
// //
// A tokio::sync::Barrier synchronizes both connections so they both have a // 1. tx1 opens a transaction and inserts revision 102. The BEFORE INSERT
// live transaction and are ready to INSERT before either proceeds. After the // trigger acquires `pg_advisory_xact_lock(lock_key)` and completes the
// barrier both race to acquire the advisory lock; one wins, commits, and // INSERT — tx1 now holds the advisory lock until it commits.
// releases the lock; the other then sees the committed MAX and is rejected // 2. tx2 opens a transaction on a second backend and issues INSERT for
// by the guard with 23514. // revision 103. The trigger fires and blocks inside
// `pg_advisory_xact_lock(lock_key)` waiting for tx1 to release.
// 3. We observe tx2's backend entering a Lock-wait state via
// pg_stat_activity (wait_event_type='Lock', wait_event='advisory'),
// with a bounded timeout — not a sleep. If the advisory-lock call is
// removed from the guard, the trigger returns immediately; tx2 never
// enters the advisory wait, and the poll times out, failing the test.
// This is the mutation-sensitivity guarantee.
// 4. tx1 commits, releasing the advisory lock. tx2 unblocks, its trigger
// reads the fresh MAX=102, and the INSERT succeeds (103 > 102).
// 5. tx2 commits. Both revisions 102 and 103 are present.
use std::time::Instant;
// tx1: open a transaction and insert revision 102. The INSERT returns after
// the trigger acquires the lock and succeeds; the advisory lock stays held
// until the transaction commits.
let mut conn1 = pool.acquire().await.expect("acquire conn1");
sqlx::query("BEGIN")
.execute(&mut *conn1)
.await
.expect("begin tx1");
sqlx::query(
"INSERT INTO identity_enrollment_policies \
(community_id, policy_revision, enrollment_mode, policy_digest, effective_at) \
VALUES ($1, 102, 1, $2, '2029-01-01T00:00:00Z')",
)
.bind(community_id)
.bind(vec![0xB1_u8; 32])
.execute(&mut *conn1)
.await
.expect("tx1 INSERT revision 102 must succeed");
// tx1 holds the advisory lock. Do NOT commit yet.
// tx2: acquire a separate backend, record its PID, then issue the INSERT.
// The trigger will block on the advisory lock held by tx1.
let pool2 = pool.clone(); let pool2 = pool.clone();
let pool3 = pool.clone(); let pool3 = pool.clone();
let community_id2 = community_id; let (pid_tx, pid_rx) = tokio::sync::oneshot::channel::<i32>();
let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(2)); let tx2_task = tokio::spawn(async move {
let barrier2 = barrier.clone(); let mut conn2 = pool2.acquire().await.expect("acquire conn2");
// Report this backend's PID so the observer can poll pg_stat_activity.
let backend_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
.fetch_one(&mut *conn2)
.await
.expect("get conn2 backend pid");
let _ = pid_tx.send(backend_pid);
sqlx::query("BEGIN")
.execute(&mut *conn2)
.await
.expect("begin tx2");
// This INSERT will block inside the trigger waiting for tx1's advisory lock.
let insert_r = sqlx::query(
"INSERT INTO identity_enrollment_policies \
(community_id, policy_revision, enrollment_mode, policy_digest, effective_at) \
VALUES ($1, 103, 1, $2, '2029-06-01T00:00:00Z')",
)
.bind(community_id)
.bind(vec![0xB2_u8; 32])
.execute(&mut *conn2)
.await;
let commit_r = sqlx::query("COMMIT").execute(&mut *conn2).await;
(insert_r, commit_r)
});
let (tx1_result, tx2_result) = tokio::join!( // Receive tx2's backend PID and wait until it enters an advisory-lock wait.
tokio::spawn(async move { // Mutation proof: without pg_advisory_xact_lock in the guard, the trigger
let mut conn = pool.acquire().await.expect("acquire conn1"); // returns immediately; tx2 never parks on an advisory lock; the poll below
sqlx::query("BEGIN") // times out and panics, making this test deterministically red.
.execute(&mut *conn) let tx2_pid = pid_rx.await.expect("tx2 reports its backend pid");
.await let deadline = Instant::now() + std::time::Duration::from_secs(10);
.expect("begin1"); loop {
// Wait until both connections have open transactions before let waiting: bool = sqlx::query_scalar(
// either races to INSERT — eliminates ordering accidents. "SELECT EXISTS (\
barrier.wait().await; SELECT 1 FROM pg_stat_activity \
let insert_r = sqlx::query( WHERE pid = $1 \
"INSERT INTO identity_enrollment_policies \ AND wait_event_type = 'Lock' \
(community_id, policy_revision, enrollment_mode, policy_digest, effective_at) \ AND wait_event = 'advisory'\
VALUES ($1, 102, 1, $2, '2029-01-01T00:00:00Z')", )",
) )
.bind(community_id) .bind(tx2_pid)
.bind(vec![0xB1_u8; 32]) .fetch_one(&pool3)
.execute(&mut *conn) .await
.await; .expect("poll tx2 advisory-lock wait");
let commit_r = sqlx::query("COMMIT").execute(&mut *conn).await; if waiting {
(insert_r, commit_r) break;
}), }
tokio::spawn(async move { assert!(
let mut conn = pool2.acquire().await.expect("acquire conn2"); Instant::now() < deadline,
sqlx::query("BEGIN") "tx2 never entered advisory-lock wait — pg_advisory_xact_lock \
.execute(&mut *conn) must be present in the guard for the lock to serialize writers"
.await );
.expect("begin2"); tokio::time::sleep(std::time::Duration::from_millis(25)).await;
// Mirror barrier wait so both are in-flight simultaneously. }
barrier2.wait().await;
let insert_r = sqlx::query(
"INSERT INTO identity_enrollment_policies \
(community_id, policy_revision, enrollment_mode, policy_digest, effective_at) \
VALUES ($1, 102, 1, $2, '2029-01-01T00:00:00Z')",
)
.bind(community_id2)
.bind(vec![0xB2_u8; 32])
.execute(&mut *conn)
.await;
let commit_r = sqlx::query("COMMIT").execute(&mut *conn).await;
(insert_r, commit_r)
}),
);
let (insert1, commit1) = tx1_result.expect("tx1 task completed"); // tx2 is observably blocked. Commit tx1, releasing the advisory lock.
let (insert2, commit2) = tx2_result.expect("tx2 task completed"); sqlx::query("COMMIT")
.execute(&mut *conn1)
.await
.expect("tx1 COMMIT must succeed");
// The BEFORE INSERT trigger fires at statement time: the winner's INSERT // tx2 unblocks: the trigger re-runs its SELECT MAX, sees committed 102,
// succeeds (lock acquired, MAX check passes, INSERT completes), the loser's // and INSERT 103 succeeds. Both the INSERT and COMMIT must complete.
// INSERT blocks waiting for the lock and then fails with 23514 when it sees let (insert2, commit2) = tx2_task.await.expect("tx2 task completed");
// the winner's committed MAX. Exactly one INSERT must succeed; exactly one insert2.expect("tx2 INSERT revision 103 must succeed after tx1 commits");
// must fail with 23514 from the named guard. commit2.expect("tx2 COMMIT must succeed");
let (i1_ok, i2_ok) = (insert1.is_ok(), insert2.is_ok());
assert!(
i1_ok ^ i2_ok,
"exactly one of the two concurrent revision-102 inserts must succeed at INSERT; \
got insert1={i1_ok} insert2={i2_ok}"
);
let loser_insert = if i1_ok { insert2 } else { insert1 };
let loser_err = loser_insert.expect_err("loser INSERT must have failed");
assert!(
loser_err
.as_database_error()
.map(|e| e.code().as_deref() == Some("23514"))
.unwrap_or(false),
"loser must fail with check_violation (23514) from \
identity_enrollment_policy_revision_monotonic guard, not a PK \
collision — this proves the advisory lock serialized the writers; \
got: {loser_err}"
);
// The winner's commit must succeed. // Both revisions 102 and 103 must be present (total: 1, 2, 100, 101, 102, 103).
let winner_commit = if i1_ok { commit1 } else { commit2 };
winner_commit.expect("winner COMMIT must succeed");
// Confirm exactly 5 policy revisions are now present (1, 2, 100, 101, 102).
let count: i64 = sqlx::query_scalar( let count: i64 = sqlx::query_scalar(
"SELECT count(*) FROM identity_enrollment_policies WHERE community_id = $1", "SELECT count(*) FROM identity_enrollment_policies WHERE community_id = $1",
) )
.bind(community_id) .bind(community_id)
.fetch_one(&pool3) .fetch_one(&pool)
.await .await
.expect("count persisted policy revisions"); .expect("count persisted policy revisions");
assert_eq!( assert_eq!(
count, 5, count, 6,
"exactly five revisions must persist after concurrency race" "exactly six revisions must persist after the controlled concurrency sequence"
); );
} }